From 2d1e8ff8cbd8cee48b2c537bab768415b1d2e055 Mon Sep 17 00:00:00 2001 From: benpankow Date: Thu, 23 Jul 2026 09:57:35 -0700 Subject: [PATCH 1/4] feat(eve): add ability to unsubscribe eve from Slack threads Signed-off-by: benpankow --- .changeset/slack-thread-unsubscribe.md | 5 + docs/channels/slack.mdx | 84 +++++- packages/eve/src/channel/routes.ts | 2 + packages/eve/src/channel/send.test.ts | 20 ++ packages/eve/src/channel/send.ts | 5 +- packages/eve/src/channel/types.ts | 2 + packages/eve/src/execution/workflow-entry.ts | 1 + packages/eve/src/execution/workflow-steps.ts | 43 ++- .../eve/src/public/channels/slack/index.ts | 3 + .../src/public/channels/slack/interactions.ts | 2 +- .../channels/slack/slackChannel.test.ts | 283 +++++++++++++++++- .../src/public/channels/slack/slackChannel.ts | 187 ++++++++++-- pnpm-lock.yaml | 21 +- 13 files changed, 603 insertions(+), 55 deletions(-) create mode 100644 .changeset/slack-thread-unsubscribe.md 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 From 59a54e5d39e5bd7388ffaf0616505cd963346cd2 Mon Sep 17 00:00:00 2001 From: benpankow Date: Mon, 27 Jul 2026 14:20:29 -0700 Subject: [PATCH 2/4] docs Signed-off-by: benpankow --- docs/channels/slack.mdx | 25 ++++++------------- .../channels/slack/slackChannel.test.ts | 17 ++++++++++++- .../src/public/channels/slack/slackChannel.ts | 11 ++++++++ 3 files changed, 35 insertions(+), 18 deletions(-) diff --git a/docs/channels/slack.mdx b/docs/channels/slack.mdx index 807ac786a..8eae234ef 100644 --- a/docs/channels/slack.mdx +++ b/docs/channels/slack.mdx @@ -73,6 +73,8 @@ Only the first available handler runs. A message hook returning `null` drops the #### Participate in active threads +> **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`. + The API grows from mention-only behavior to automatic or stateful participation. **1. Respond only to explicit mentions.** This is the default: @@ -92,8 +94,6 @@ export default slackChannel({ }); ``` -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" @@ -118,42 +118,33 @@ export default slackChannel({ 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: +**4. Classify ambiguous replies.** A handler can use a model to decide whether the current message needs a response: ```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, + initialState: null, async handle(message, ctx) { - if (ctx.isBotMentioned) return { respond: true, state: "followingThread" }; - if (ctx.state === "ignoringThread") return { respond: false, state: ctx.state }; + if (ctx.isBotMentioned) return { respond: true, state: null }; - 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, + system: "Decide whether this Slack message expects the agent to answer.", + prompt: message.text, output: Output.object({ schema: z.object({ respond: z.boolean() }) }), timeout: 1_500, maxRetries: 0, }); - return { respond: output.respond, state: ctx.state }; + return { respond: output.respond, state: null }; }, }, }); ``` -`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/public/channels/slack/slackChannel.test.ts b/packages/eve/src/public/channels/slack/slackChannel.test.ts index 049938260..65ea3b6a9 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.test.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.test.ts @@ -1572,7 +1572,12 @@ describe("slackChannel() onMessage", () => { }); it("resolves thread participation inside the hydrated session", async () => { - const participate = vi.fn(() => ({ respond: false, state: "ignoringThread" })); + const log = vi.spyOn(console, "log").mockImplementation(() => {}); + const participate = vi.fn(() => ({ + reason: "human-handoff", + respond: false, + state: "ignoringThread", + })); const channel = slackChannel({ threadParticipation: { initialState: "followingThread", handle: participate }, }); @@ -1608,6 +1613,16 @@ describe("slackChannel() onMessage", () => { expect(result).toBeUndefined(); expect(adapterCtx.state.participationState).toBe("ignoringThread"); + expect(log).toHaveBeenCalledWith( + "[eve:slack.channel] Slack thread message skipped by thread participation", + { + channelId: "C01", + messageTs: "1700000000.000002", + reason: "human-handoff", + sessionId: "test-session", + threadTs: "1700000000.000001", + }, + ); expect(participate).toHaveBeenCalledWith( message, expect.objectContaining({ diff --git a/packages/eve/src/public/channels/slack/slackChannel.ts b/packages/eve/src/public/channels/slack/slackChannel.ts index 3b85189b0..142fabc27 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.ts @@ -213,6 +213,8 @@ export interface SlackThreadParticipationContext extends SessionContext export interface SlackThreadParticipationResult { /** Whether this message should dispatch an agent turn. */ readonly respond: boolean; + /** Optional short observability label. Do not include message content or secrets. */ + readonly reason?: string; /** JSON-serializable state made available to the next thread reply. */ readonly state: TState; } @@ -817,6 +819,15 @@ export function slackChannel( ); } adapterCtx.state.participationState = participation.state; + if (!participation.respond) { + log.info("Slack thread message skipped by thread participation", { + channelId: delivery.message.channelId, + messageTs: delivery.message.ts, + reason: participation.reason, + sessionId: sessionCtx.session.id, + threadTs: delivery.message.threadTs, + }); + } return participation.respond ? defaultDeliverResult(payload) : undefined; }, fetchFile: slackFetchFile, From 43d6f6fba61524117abd64e04a79da384e94470e Mon Sep 17 00:00:00 2001 From: benpankow Date: Mon, 27 Jul 2026 14:50:31 -0700 Subject: [PATCH 3/4] renames Signed-off-by: benpankow --- .changeset/slack-thread-replies.md | 5 + .changeset/slack-thread-unsubscribe.md | 5 - docs/channels/slack.mdx | 31 ++--- .../eve/src/public/channels/slack/index.ts | 6 +- .../channels/slack/slackChannel.test.ts | 76 +++++------ .../src/public/channels/slack/slackChannel.ts | 128 ++++++++---------- 6 files changed, 116 insertions(+), 135 deletions(-) create mode 100644 .changeset/slack-thread-replies.md delete mode 100644 .changeset/slack-thread-unsubscribe.md diff --git a/.changeset/slack-thread-replies.md b/.changeset/slack-thread-replies.md new file mode 100644 index 000000000..1fe29cb61 --- /dev/null +++ b/.changeset/slack-thread-replies.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Added Slack `threadReplies` configuration for responding to replies in active threads with arbitrary durable user-defined state. `onReply` independently decides whether to respond and which JSON-serializable state is available to the next thread reply. diff --git a/.changeset/slack-thread-unsubscribe.md b/.changeset/slack-thread-unsubscribe.md deleted file mode 100644 index 75e2f98fb..000000000 --- a/.changeset/slack-thread-unsubscribe.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"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 8eae234ef..b6115e83d 100644 --- a/docs/channels/slack.mdx +++ b/docs/channels/slack.mdx @@ -56,7 +56,7 @@ VERCEL_USE_EXPERIMENTAL_FRAMEWORKS=1 vercel deploy --prod Message hooks return `{ auth }` to dispatch, `null` to drop, or `{ auth, context }` to inject background into history. -- `onMessage(ctx, message)` handles Slack `message` events. eve drops messages authored by the installed app before this hook runs, preventing self-reply loops. Messages from other bots remain visible; check `message.author?.isBot` when those should also be ignored. `ctx.isBotMentioned()` and `ctx.isSubscribed()` support mention and active-thread policies. +- `onMessage(ctx, message)` handles Slack `message` events. eve drops messages authored by the installed app before this hook runs, preventing self-reply loops. Messages from other bots remain visible; check `message.author?.isBot` when those should also be ignored. `ctx.isBotMentioned()` identifies explicit mentions. - `onAppMention(ctx, message)` handles only `app_mention` and takes precedence over `onMessage`. Its default derives workspace-scoped auth and posts `Thinking…`. - `onDirectMessage(ctx, message)` handles only DMs and takes precedence over `onMessage`. Bot-authored messages and edits are filtered first; Slack requires `message.im` and `im:history`. @@ -71,11 +71,11 @@ 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. -#### Participate in active threads +#### Respond to active thread replies > **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`. -The API grows from mention-only behavior to automatic or stateful participation. +By default, eve responds only to explicit mentions. `threadReplies` adds automatic or stateful response decisions for replies in a thread with an active eve session. **1. Respond only to explicit mentions.** This is the default: @@ -85,25 +85,25 @@ export default slackChannel({ }); ``` -**2. Respond to every human reply in active threads.** Enable `threadParticipation` without requiring another mention: +**2. Respond to every human reply in active threads.** Enable `threadReplies` without requiring another mention: ```ts title="agent/channels/slack.ts" export default slackChannel({ credentials: connectSlackCredentials("slack/my-agent"), - threadParticipation: true, + threadReplies: true, }); ``` -**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: +**3. Track your own reply state.** Provide an `initialState` and return the next state from `onReply`. This example treats mentioning another person as a handoff: ```ts title="agent/channels/slack.ts" -type ParticipationState = "followingThread" | "ignoringThread"; +type ThreadReplyState = "followingThread" | "ignoringThread"; export default slackChannel({ credentials: connectSlackCredentials("slack/my-agent"), - threadParticipation: { - initialState: "followingThread" as ParticipationState, - handle(message, ctx) { + threadReplies: { + initialState: "followingThread" as ThreadReplyState, + onReply(message, ctx) { if (ctx.isBotMentioned) return { respond: true, state: "followingThread" }; if (ctx.state === "ignoringThread") return { respond: false, state: ctx.state }; @@ -116,7 +116,7 @@ export default slackChannel({ }); ``` -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. +Thread replies pass through two phases. First, the webhook admits only human replies that belong to an active eve session; `onMessage`, when configured, replaces that default admission policy and can apply custom auth or pre-dispatch work. Second, `threadReplies.onReply` runs inside the hydrated durable session and decides whether the admitted reply dispatches an agent turn. It receives thread replies, including explicit mentions, but not the top-level message that starts the thread; that message initializes `initialState`. `state` must be JSON-serializable. Returning `respond: false` skips this reply but leaves the session active, so the policy runs again for a later reply. **4. Classify ambiguous replies.** A handler can use a model to decide whether the current message needs a response: @@ -126,9 +126,9 @@ import { z } from "zod"; export default slackChannel({ credentials: connectSlackCredentials("slack/my-agent"), - threadParticipation: { + threadReplies: { initialState: null, - async handle(message, ctx) { + async onReply(message, ctx) { if (ctx.isBotMentioned) return { respond: true, state: null }; const { output } = await generateText({ @@ -145,7 +145,7 @@ export default slackChannel({ }); ``` -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 `onMessage` only when you need custom webhook admission, auth, or pre-dispatch behavior. It is the first phase and runs before durable session state is hydrated; `threadReplies.onReply` is the second phase. `ctx.isBotMentioned()` identifies explicit mentions. 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: @@ -171,8 +171,7 @@ Message hooks (`onMessage`, `onAppMention`, and `onDirectMessage`) receive a thr export default slackChannel({ credentials: connectSlackCredentials("slack/my-agent"), async onMessage(ctx, message) { - const shouldHandle = ctx.isBotMentioned() || (await ctx.isSubscribed()); - if (!shouldHandle || message.author?.isBot) return null; + if (!ctx.isBotMentioned() || message.author?.isBot) return null; await ctx.cancel(); return { auth: null }; diff --git a/packages/eve/src/public/channels/slack/index.ts b/packages/eve/src/public/channels/slack/index.ts index 6a350feaa..8192e8ef8 100644 --- a/packages/eve/src/public/channels/slack/index.ts +++ b/packages/eve/src/public/channels/slack/index.ts @@ -39,9 +39,9 @@ export { type SlackResetOptions, type SlackSessionTarget, type SlackThread, - type SlackThreadParticipation, - type SlackThreadParticipationContext, - type SlackThreadParticipationResult, + type SlackThreadReplies, + type SlackThreadReplyContext, + type SlackThreadReplyResult, type SlackWebhookVerifier, type SlackWorkspaceHandle, } from "#public/channels/slack/slackChannel.js"; diff --git a/packages/eve/src/public/channels/slack/slackChannel.test.ts b/packages/eve/src/public/channels/slack/slackChannel.test.ts index 65ea3b6a9..71771455a 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.test.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.test.ts @@ -1433,13 +1433,10 @@ describe("slackChannel() inbound mention pipeline", () => { }); describe("slackChannel() onMessage", () => { - it("exposes mention and active-session helpers", async () => { - const observed: Array<{ mentioned: boolean; subscribed: boolean }> = []; + it("exposes the mention helper", async () => { + const observed: boolean[] = []; const onMessage = vi.fn(async (ctx) => { - observed.push({ - mentioned: ctx.isBotMentioned(), - subscribed: await ctx.isSubscribed(), - }); + observed.push(ctx.isBotMentioned()); return null; }); const channel = slackChannel({ @@ -1464,10 +1461,7 @@ describe("slackChannel() onMessage", () => { ); await firePost(channel, buildSignedRequest({ body: reply })); - expect(observed).toEqual([ - { mentioned: true, subscribed: true }, - { mentioned: false, subscribed: true }, - ]); + expect(observed).toEqual([true, false]); }); it("ignores ordinary thread replies by default", async () => { @@ -1492,10 +1486,10 @@ describe("slackChannel() onMessage", () => { expect(send).not.toHaveBeenCalled(); }); - it("does not start a session from an ordinary reply when threadParticipation is enabled", async () => { + it("does not start a session from an ordinary reply when threadReplies is enabled", async () => { const channel = slackChannel({ credentials: { botToken: "xoxb-test", signingSecret: SIGNING_SECRET }, - threadParticipation: true, + threadReplies: true, }); const reply = buildEventBody( { @@ -1517,10 +1511,10 @@ describe("slackChannel() onMessage", () => { expect(send).not.toHaveBeenCalled(); }); - it("follows active threads when threadParticipation is enabled", async () => { + it("responds to replies in active threads when threadReplies is enabled", async () => { const channel = slackChannel({ credentials: { botToken: "xoxb-test", signingSecret: SIGNING_SECRET }, - threadParticipation: true, + threadReplies: true, }); const reply = buildEventBody( { @@ -1546,9 +1540,7 @@ describe("slackChannel() onMessage", () => { it("allows an advanced onMessage admission policy", async () => { const channel = slackChannel({ credentials: { botToken: "xoxb-test", signingSecret: SIGNING_SECRET }, - async onMessage(ctx) { - return ctx.isBotMentioned() || (await ctx.isSubscribed()) ? { auth: null } : null; - }, + onMessage: () => ({ auth: null }), }); const reply = buildEventBody( { @@ -1571,19 +1563,19 @@ describe("slackChannel() onMessage", () => { ); }); - it("resolves thread participation inside the hydrated session", async () => { + it("resolves thread replies inside the hydrated session", async () => { const log = vi.spyOn(console, "log").mockImplementation(() => {}); - const participate = vi.fn(() => ({ + const onReply = vi.fn(() => ({ reason: "human-handoff", respond: false, state: "ignoringThread", })); const channel = slackChannel({ - threadParticipation: { initialState: "followingThread", handle: participate }, + threadReplies: { initialState: "followingThread", onReply }, }); const adapter = withState(getAdapter(channel), { ...THREAD_STATE, - participationState: "followingThread", + threadReplyState: "followingThread", }); const adapterCtx = buildAdapterContext(adapter, stubAccessor()); const message = { @@ -1612,9 +1604,9 @@ describe("slackChannel() onMessage", () => { ); expect(result).toBeUndefined(); - expect(adapterCtx.state.participationState).toBe("ignoringThread"); + expect(adapterCtx.state.threadReplyState).toBe("ignoringThread"); expect(log).toHaveBeenCalledWith( - "[eve:slack.channel] Slack thread message skipped by thread participation", + "[eve:slack.channel] Slack thread message skipped by thread reply policy", { channelId: "C01", messageTs: "1700000000.000002", @@ -1623,7 +1615,7 @@ describe("slackChannel() onMessage", () => { threadTs: "1700000000.000001", }, ); - expect(participate).toHaveBeenCalledWith( + expect(onReply).toHaveBeenCalledWith( message, expect.objectContaining({ state: "followingThread", @@ -1635,16 +1627,16 @@ describe("slackChannel() onMessage", () => { ); }); - it("can ignore one message without changing participation state", async () => { + it("can ignore one message without changing reply state", async () => { const channel = slackChannel({ - threadParticipation: { + threadReplies: { initialState: "followingThread", - handle: () => ({ respond: false, state: "followingThread" }), + onReply: () => ({ respond: false, state: "followingThread" }), }, }); const adapter = withState(getAdapter(channel), { ...THREAD_STATE, - participationState: "followingThread", + threadReplyState: "followingThread", }); const adapterCtx = buildAdapterContext(adapter, stubAccessor()); const message = { @@ -1673,17 +1665,17 @@ describe("slackChannel() onMessage", () => { ); expect(result).toBeUndefined(); - expect(adapterCtx.state.participationState).toBe("followingThread"); + expect(adapterCtx.state.threadReplyState).toBe("followingThread"); }); - it("does not run thread participation for a top-level session message", async () => { - const participate = vi.fn(() => ({ respond: false, state: "ignoringThread" })); + it("does not run thread reply policy for a top-level session message", async () => { + const onReply = vi.fn(() => ({ respond: false, state: "ignoringThread" })); const channel = slackChannel({ - threadParticipation: { initialState: "followingThread", handle: participate }, + threadReplies: { initialState: "followingThread", onReply }, }); const adapter = withState(getAdapter(channel), { ...THREAD_STATE, - participationState: "ignoringThread", + threadReplyState: "ignoringThread", }); const adapterCtx = buildAdapterContext(adapter, stubAccessor()); const message = { @@ -1711,19 +1703,19 @@ describe("slackChannel() onMessage", () => { ), ); - expect(participate).not.toHaveBeenCalled(); + expect(onReply).not.toHaveBeenCalled(); expect(result).toEqual({ message: "attributed message" }); - expect(adapterCtx.state.participationState).toBe("followingThread"); + expect(adapterCtx.state.threadReplyState).toBe("followingThread"); }); - it("runs thread participation for explicit mentions in a thread", async () => { - const participate = vi.fn(() => ({ respond: true, state: "followingThread" })); + it("runs thread reply policy for explicit mentions in a thread", async () => { + const onReply = vi.fn(() => ({ respond: true, state: "followingThread" })); const channel = slackChannel({ - threadParticipation: { initialState: "followingThread", handle: participate }, + threadReplies: { initialState: "followingThread", onReply }, }); const adapter = withState(getAdapter(channel), { ...THREAD_STATE, - participationState: "ignoringThread", + threadReplyState: "ignoringThread", }); const adapterCtx = buildAdapterContext(adapter, stubAccessor()); const message = { @@ -1751,7 +1743,7 @@ describe("slackChannel() onMessage", () => { ), ); - expect(participate).toHaveBeenCalledWith( + expect(onReply).toHaveBeenCalledWith( message, expect.objectContaining({ state: "ignoringThread", isBotMentioned: true }), ); @@ -1760,7 +1752,7 @@ describe("slackChannel() onMessage", () => { it("responds to explicit mentions in a thread by default", async () => { const adapter = withState(getAdapter(slackChannel()), { ...THREAD_STATE, - participationState: "ignoringThread", + threadReplyState: "ignoringThread", }); const adapterCtx = buildAdapterContext(adapter, stubAccessor()); const message = { @@ -3160,7 +3152,7 @@ describe("constrainAuthorizationRequired", () => { const request = vi.fn().mockResolvedValue({ ok: true }); const state: SlackChannelState = { channelId: "C123", - participationState: "followingThread", + threadReplyState: "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 142fabc27..bf92eea7c 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.ts @@ -166,18 +166,10 @@ type SlackSessionFailedHandler = ( channel: SlackEventContext, ) => void | Promise; -function defaultThreadParticipation(): SlackThreadParticipationResult { +function defaultThreadReplies(): SlackThreadReplyResult { 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; @@ -197,8 +189,8 @@ function readSlackMessageDeliveryData(value: unknown): SlackMessageDeliveryData * 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 { +/** Context for a durable Slack thread reply policy. */ +export interface SlackThreadReplyContext extends SessionContext { /** Durable user-defined state returned after the previous thread reply. */ readonly state: TState; /** Whether this message explicitly mentions the installed bot. */ @@ -210,7 +202,7 @@ export interface SlackThreadParticipationContext extends SessionContext } /** Response decision and durable user-defined state after one thread reply. */ -export interface SlackThreadParticipationResult { +export interface SlackThreadReplyResult { /** Whether this message should dispatch an agent turn. */ readonly respond: boolean; /** Optional short observability label. Do not include message content or secrets. */ @@ -219,15 +211,15 @@ export interface SlackThreadParticipationResult { readonly state: TState; } -/** Custom durable Slack thread participation behavior. */ -export interface SlackThreadParticipation { +/** Custom durable policy for deciding whether to respond to thread replies. */ +export interface SlackThreadReplies { /** 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( + onReply( message: SlackMessage, - ctx: SlackThreadParticipationContext, - ): SlackThreadParticipationResult | Promise>; + ctx: SlackThreadReplyContext, + ): SlackThreadReplyResult | Promise>; } interface SlackMessageDeliveryData { @@ -237,8 +229,8 @@ interface SlackMessageDeliveryData { } export interface SlackChannelState { - /** User-defined durable state for thread participation. */ - participationState?: unknown; + /** User-defined durable state for thread reply policy. */ + threadReplyState?: unknown; /** Slack channel id seeded by the inbound mention. */ channelId: string | null; /** Slack thread root ts. */ @@ -424,11 +416,6 @@ 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. - * This webhook-side lookup does not read the durable subscription state. - */ - isSubscribed(): Promise; /** Returns whether the inbound event explicitly mentions this bot. */ isBotMentioned(): boolean; /** @@ -561,7 +548,7 @@ export interface SlackChannelInternalEvents extends Omit< readonly "authorization.required"?: SlackEventHandler<"authorization.required">; } -export interface SlackChannelConfig { +export interface SlackChannelConfig { readonly credentials?: SlackChannelCredentials; readonly botName?: string; @@ -586,13 +573,13 @@ 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. + * Responds to replies in threads with an active eve session. Pass `true` to + * respond to every admitted human reply, or provide durable user-defined + * state and an `onReply` policy that decides whether to respond. The policy + * runs after Slack channel state and `defineState` are hydrated; top-level + * messages bypass it and initialize the configured state. */ - readonly threadParticipation?: true | SlackThreadParticipation; + readonly threadReplies?: true | SlackThreadReplies; /** * Handles human-authored Slack messages. Specialized `onAppMention` and @@ -743,8 +730,8 @@ function defineSlackChannel( * fallback ahead of unsupplied mention and DM defaults; otherwise unsupplied * fields keep their defaults. */ -export function slackChannel( - config: SlackChannelConfig = {}, +export function slackChannel( + config: SlackChannelConfig = {}, ): SlackChannel { const uploadPolicy = mergeUploadPolicy(config.uploadPolicy); const slackFetchFile = createSlackFetchFile({ botToken: config.credentials?.botToken }); @@ -781,10 +768,10 @@ export function slackChannel( pendingToolCallMessage: null, lastReasoningTypingAtMs: null, lastReasoningTypingStatus: null, - participationState: - typeof config.threadParticipation === "object" - ? structuredClone(config.threadParticipation.initialState) - : config.threadParticipation === true + threadReplyState: + typeof config.threadReplies === "object" + ? structuredClone(config.threadReplies.initialState) + : config.threadReplies === true ? true : undefined, pendingAuthMessageTs: {}, @@ -793,42 +780,40 @@ export function slackChannel( 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, - ); + if (typeof config.threadReplies === "object") { + adapterCtx.state.threadReplyState = structuredClone(config.threadReplies.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, { + const reply = + typeof config.threadReplies === "object" + ? await config.threadReplies.onReply(delivery.message, { ...sessionCtx, isBotMentioned: delivery.isBotMentioned, slack: slackCtx.slack, - state: adapterCtx.state.participationState as TParticipationState, + state: adapterCtx.state.threadReplyState as TThreadReplyState, thread: slackCtx.thread, }) - : defaultThreadParticipation(); - if (typeof participation?.respond !== "boolean" || !("state" in participation)) { + : defaultThreadReplies(); + if (typeof reply?.respond !== "boolean" || !("state" in reply)) { throw new Error( - "slackChannel().threadParticipation.handle must return `respond` and `state` fields.", + "slackChannel().threadReplies.onReply must return `respond` and `state` fields.", ); } - adapterCtx.state.participationState = participation.state; - if (!participation.respond) { - log.info("Slack thread message skipped by thread participation", { + adapterCtx.state.threadReplyState = reply.state; + if (!reply.respond) { + log.info("Slack thread message skipped by thread reply policy", { channelId: delivery.message.channelId, messageTs: delivery.message.ts, - reason: participation.reason, + reason: reply.reason, sessionId: sessionCtx.session.id, threadTs: delivery.message.threadTs, }); } - return participation.respond ? defaultDeliverResult(payload) : undefined; + return reply.respond ? defaultDeliverResult(payload) : undefined; }, fetchFile: slackFetchFile, metadata(state): SlackInstrumentationMetadata { @@ -1066,10 +1051,7 @@ async function handleEventPost(input: { } } - if ( - dispatch === null && - (config.onMessage !== undefined || config.threadParticipation !== undefined) - ) { + if (dispatch === null && (config.onMessage !== undefined || config.threadReplies !== undefined)) { const message = parseMessageEvent(envelope); if (message !== null && !isSelfAuthoredSlackMessage({ appId, botUserId }, message)) { // Slack also emits message.channels for an app mention. The app_mention @@ -1081,7 +1063,7 @@ async function handleEventPost(input: { botUserId, cancel: input.cancel, credentials: config.credentials, - handler: config.onMessage ?? defaultSubscribedThreadMessage, + handler: config.onMessage, kind: "channel_message", message, resolveActiveSession: input.resolveActiveSession, @@ -1149,7 +1131,7 @@ async function dispatchSlackMessage(input: { readonly botUserId: string | undefined; readonly cancel: CancelFn; readonly credentials: SlackChannelCredentials | undefined; - readonly handler: NonNullable["onMessage"]>; + readonly handler?: SlackChannelConfig["onMessage"]; readonly kind: "app_mention" | "channel_message" | "direct_message"; readonly message: SlackMessage; readonly resolveActiveSession: (options: { @@ -1179,8 +1161,6 @@ async function dispatchSlackMessage(input: { turnId: options.turnId, }), isBotMentioned: () => isBotMentioned, - isSubscribed: async () => - (await input.resolveActiveSession({ continuationToken })) !== undefined, reset: (options = {}) => input.reset({ continuationToken, @@ -1190,14 +1170,24 @@ async function dispatchSlackMessage(input: { thread, }; - let result; - try { - result = await input.handler(ctx, input.message); - } catch (error) { - logError(log, `${input.kind} handler failed`, error, { - channelId: input.message.channelId, - }); - return; + let result: SlackInboundResult | undefined; + if (input.handler === undefined) { + if ( + input.message.author?.isBot || + (await input.resolveActiveSession({ continuationToken })) === undefined + ) { + return; + } + result = { auth: defaultSlackAuth(input.message, ctx) }; + } else { + try { + result = await input.handler(ctx, input.message); + } catch (error) { + logError(log, `${input.kind} handler failed`, error, { + channelId: input.message.channelId, + }); + return; + } } if (result === null || result === undefined) return; From d0b1c1d7a25a9da7d997fc571683de6b226c081f Mon Sep 17 00:00:00 2001 From: benpankow Date: Mon, 27 Jul 2026 14:55:43 -0700 Subject: [PATCH 4/4] rename2 Signed-off-by: benpankow --- docs/channels/slack.mdx | 9 +- .../eve/src/public/channels/slack/index.ts | 4 + .../channels/slack/slackChannel.test.ts | 38 +++++++++ .../src/public/channels/slack/slackChannel.ts | 83 +++++++++++++------ 4 files changed, 103 insertions(+), 31 deletions(-) diff --git a/docs/channels/slack.mdx b/docs/channels/slack.mdx index b6115e83d..84f19b0d2 100644 --- a/docs/channels/slack.mdx +++ b/docs/channels/slack.mdx @@ -116,9 +116,9 @@ export default slackChannel({ }); ``` -Thread replies pass through two phases. First, the webhook admits only human replies that belong to an active eve session; `onMessage`, when configured, replaces that default admission policy and can apply custom auth or pre-dispatch work. Second, `threadReplies.onReply` runs inside the hydrated durable session and decides whether the admitted reply dispatches an agent turn. It receives thread replies, including explicit mentions, but not the top-level message that starts the thread; that message initializes `initialState`. `state` must be JSON-serializable. Returning `respond: false` skips this reply but leaves the session active, so the policy runs again for a later reply. +Thread replies pass through two phases. First, the webhook admits only human replies that belong to an active eve session; `onMessage`, when configured, replaces that default admission policy and can apply custom auth or pre-dispatch work. Second, `threadReplies.onReply` runs inside the hydrated durable session and decides whether the admitted reply dispatches an agent turn. It receives thread replies, including explicit mentions, but not the top-level message that starts the thread; that message initializes `initialState` when the policy uses state. Stateful `state` values must be JSON-serializable. Returning `respond: false` skips this reply but leaves the session active, so the policy runs again for a later reply. -**4. Classify ambiguous replies.** A handler can use a model to decide whether the current message needs a response: +**4. Classify ambiguous replies without state.** Omit `initialState` when the policy does not need durable state. `onReply` then returns only its response decision: ```ts title="agent/channels/slack.ts" import { generateText, Output } from "ai"; @@ -127,9 +127,8 @@ import { z } from "zod"; export default slackChannel({ credentials: connectSlackCredentials("slack/my-agent"), threadReplies: { - initialState: null, async onReply(message, ctx) { - if (ctx.isBotMentioned) return { respond: true, state: null }; + if (ctx.isBotMentioned) return { respond: true }; const { output } = await generateText({ model: "google/gemini-3.1-flash-lite", @@ -139,7 +138,7 @@ export default slackChannel({ timeout: 1_500, maxRetries: 0, }); - return { respond: output.respond, state: null }; + return { respond: output.respond }; }, }, }); diff --git a/packages/eve/src/public/channels/slack/index.ts b/packages/eve/src/public/channels/slack/index.ts index 8192e8ef8..b355d6244 100644 --- a/packages/eve/src/public/channels/slack/index.ts +++ b/packages/eve/src/public/channels/slack/index.ts @@ -39,6 +39,10 @@ export { type SlackResetOptions, type SlackSessionTarget, type SlackThread, + type SlackStatefulThreadReplies, + type SlackStatefulThreadReplyContext, + type SlackStatefulThreadReplyResult, + type SlackStatelessThreadReplies, type SlackThreadReplies, type SlackThreadReplyContext, type SlackThreadReplyResult, diff --git a/packages/eve/src/public/channels/slack/slackChannel.test.ts b/packages/eve/src/public/channels/slack/slackChannel.test.ts index 71771455a..a2e7f07d7 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.test.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.test.ts @@ -1627,6 +1627,44 @@ describe("slackChannel() onMessage", () => { ); }); + it("runs a stateless thread reply policy", async () => { + const onReply = vi.fn(() => ({ respond: false })); + const channel = slackChannel({ threadReplies: { onReply } }); + const adapter = withState(getAdapter(channel), THREAD_STATE); + 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.threadReplyState).toBeUndefined(); + expect(onReply).toHaveBeenCalledWith( + message, + expect.not.objectContaining({ state: expect.anything() }), + ); + }); + it("can ignore one message without changing reply state", async () => { const channel = slackChannel({ threadReplies: { diff --git a/packages/eve/src/public/channels/slack/slackChannel.ts b/packages/eve/src/public/channels/slack/slackChannel.ts index bf92eea7c..247916bed 100644 --- a/packages/eve/src/public/channels/slack/slackChannel.ts +++ b/packages/eve/src/public/channels/slack/slackChannel.ts @@ -166,7 +166,7 @@ type SlackSessionFailedHandler = ( channel: SlackEventContext, ) => void | Promise; -function defaultThreadReplies(): SlackThreadReplyResult { +function defaultThreadReplies(): SlackStatefulThreadReplyResult { return { respond: true, state: true }; } @@ -189,10 +189,8 @@ function readSlackMessageDeliveryData(value: unknown): SlackMessageDeliveryData * step boundaries. Anything written here must round-trip through * `JSON.stringify` / `JSON.parse`. */ -/** Context for a durable Slack thread reply policy. */ -export interface SlackThreadReplyContext extends SessionContext { - /** Durable user-defined state returned after the previous thread reply. */ - readonly state: TState; +/** Context for a Slack thread reply policy. */ +export interface SlackThreadReplyContext extends SessionContext { /** Whether this message explicitly mentions the installed bot. */ readonly isBotMentioned: boolean; /** Thread-bound Slack operations, including lazy history refresh. */ @@ -201,27 +199,51 @@ export interface SlackThreadReplyContext extends SessionContext { readonly slack: SlackHandle; } -/** Response decision and durable user-defined state after one thread reply. */ -export interface SlackThreadReplyResult { +/** Response decision after one thread reply. */ +export interface SlackThreadReplyResult { /** Whether this message should dispatch an agent turn. */ readonly respond: boolean; /** Optional short observability label. Do not include message content or secrets. */ readonly reason?: string; +} + +/** Context for a thread reply policy with durable user-defined state. */ +export interface SlackStatefulThreadReplyContext extends SlackThreadReplyContext { + /** Durable user-defined state returned after the previous thread reply. */ + readonly state: TState; +} + +/** Response decision and next durable state after one thread reply. */ +export interface SlackStatefulThreadReplyResult extends SlackThreadReplyResult { /** JSON-serializable state made available to the next thread reply. */ readonly state: TState; } -/** Custom durable policy for deciding whether to respond to thread replies. */ -export interface SlackThreadReplies { +/** Stateless policy for deciding whether to respond to thread replies. */ +export interface SlackStatelessThreadReplies { + /** Runs for each admitted reply after the session is hydrated. */ + onReply( + message: SlackMessage, + ctx: SlackThreadReplyContext, + ): SlackThreadReplyResult | Promise; +} + +/** Stateful policy for deciding whether to respond to thread replies. */ +export interface SlackStatefulThreadReplies { /** 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. */ onReply( message: SlackMessage, - ctx: SlackThreadReplyContext, - ): SlackThreadReplyResult | Promise>; + ctx: SlackStatefulThreadReplyContext, + ): SlackStatefulThreadReplyResult | Promise>; } +/** Policy for deciding whether to respond to thread replies. */ +export type SlackThreadReplies = + | SlackStatelessThreadReplies + | SlackStatefulThreadReplies; + interface SlackMessageDeliveryData { readonly isBotMentioned: boolean; readonly kind: "slack-message"; @@ -769,18 +791,16 @@ export function slackChannel( lastReasoningTypingAtMs: null, lastReasoningTypingStatus: null, threadReplyState: - typeof config.threadReplies === "object" + typeof config.threadReplies === "object" && "initialState" in config.threadReplies ? structuredClone(config.threadReplies.initialState) - : config.threadReplies === true - ? true - : undefined, + : 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.threadReplies === "object") { + if (typeof config.threadReplies === "object" && "initialState" in config.threadReplies) { adapterCtx.state.threadReplyState = structuredClone(config.threadReplies.initialState); } return defaultDeliverResult(payload); @@ -788,22 +808,33 @@ export function slackChannel( const sessionCtx = buildCallbackContext(); const slackCtx = adapterCtx as typeof adapterCtx & SlackChannelContext; + const stateful = + typeof config.threadReplies === "object" && "initialState" in config.threadReplies; const reply = typeof config.threadReplies === "object" - ? await config.threadReplies.onReply(delivery.message, { - ...sessionCtx, - isBotMentioned: delivery.isBotMentioned, - slack: slackCtx.slack, - state: adapterCtx.state.threadReplyState as TThreadReplyState, - thread: slackCtx.thread, - }) + ? stateful + ? await config.threadReplies.onReply(delivery.message, { + ...sessionCtx, + isBotMentioned: delivery.isBotMentioned, + slack: slackCtx.slack, + state: adapterCtx.state.threadReplyState as TThreadReplyState, + thread: slackCtx.thread, + }) + : await config.threadReplies.onReply(delivery.message, { + ...sessionCtx, + isBotMentioned: delivery.isBotMentioned, + slack: slackCtx.slack, + thread: slackCtx.thread, + }) : defaultThreadReplies(); - if (typeof reply?.respond !== "boolean" || !("state" in reply)) { + if (typeof reply?.respond !== "boolean" || (stateful && !("state" in reply))) { throw new Error( - "slackChannel().threadReplies.onReply must return `respond` and `state` fields.", + stateful + ? "slackChannel().threadReplies.onReply must return `respond` and `state` fields." + : "slackChannel().threadReplies.onReply must return a `respond` field.", ); } - adapterCtx.state.threadReplyState = reply.state; + if (stateful && "state" in reply) adapterCtx.state.threadReplyState = reply.state; if (!reply.respond) { log.info("Slack thread message skipped by thread reply policy", { channelId: delivery.message.channelId,