feat(eve): add stateful Slack threadReplies hook

.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.

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,24 +71,80 @@ 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
+#### Respond to active thread replies
 
-Use `onMessage` to handle explicit mentions and continue replies in threads with an active eve session:
+> **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`.
+
+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:
 
 ```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 `threadReplies` without requiring another mention:
+
+```ts title="agent/channels/slack.ts"
+export default slackChannel({
+  credentials: connectSlackCredentials("slack/my-agent"),
+  threadReplies: true,
+});
+```
+
+**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 ThreadReplyState = "followingThread" | "ignoringThread";
+
+export default slackChannel({
+  credentials: connectSlackCredentials("slack/my-agent"),
+  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 };
+
+      const mentionsSomeone = /<@[A-Z0-9]+(?:\|[^>]+)?>/.test(message.text);
+      return mentionsSomeone
+        ? { respond: false, state: "ignoringThread" }
+        : { respond: true, state: ctx.state };
+    },
+  },
+});
+```
+
+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 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";
+import { z } from "zod";
+
+export default slackChannel({
+  credentials: connectSlackCredentials("slack/my-agent"),
+  threadReplies: {
+    async onReply(message, ctx) {
+      if (ctx.isBotMentioned) return { respond: true };
+
+      const { output } = await generateText({
+        model: "google/gemini-3.1-flash-lite",
+        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 };
+    },
   },
 });
 ```
 
-`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`.
+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:
 
@@ -114,8 +170,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 };

packages/eve/src/channel/routes.ts

@@ -44,6 +44,8 @@ export interface RouteHandlerArgs<TState = undefined> {
 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

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(),

packages/eve/src/channel/send.ts

@@ -29,6 +29,7 @@ export function createSendFn<TState = undefined>(
     const {
       message: rawMessage,
       inputResponses,
+      channelData,
       context,
       outputSchema,
     } = normalizeSendInput(input);
@@ -39,7 +40,7 @@ export function createSendFn<TState = undefined>(
         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<TState = undefined>(
       channelName,
       callback,
       continuationToken,
-      input: { message: message ?? "", context, outputSchema },
+      input: { channelData, message: message ?? "", context, outputSchema },
       mode,
       requestId: metadata.requestId,
       title,

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;
   };

packages/eve/src/execution/workflow-entry.ts

@@ -112,6 +112,7 @@ export async function workflowEntry(input: WorkflowEntryInput): Promise<Workflow
         kind: "deliver",
         payloads: [
           {
+            channelData: input.input.channelData,
             message: input.input.message,
             context: input.input.context,
             outputSchema: input.input.outputSchema,

packages/eve/src/execution/workflow-steps.ts

@@ -16,7 +16,7 @@ import {
   SessionDynamicToolRuntimeRevisionKey,
 } from "#context/keys.js";
 import { BundleKey, ChannelKey } from "#runtime/sessions/runtime-context-keys.js";
-import { runStep } from "#context/run-step.js";
+import { runStep, withContextScope } from "#context/run-step.js";
 import { deserializeContext, serializeContext } from "#context/serialize.js";
 import { getHarnessEmissionState } from "#harness/emission.js";
 import {
@@ -209,7 +209,7 @@ export async function turnStep(rawInput: TurnStepInput): Promise<DurableStepResu
     ctx.set(AuthKey, input.input.auth ?? null);
   }
 
-  const initialSession = hydrateDurableSession({
+  let initialSession = hydrateDurableSession({
     compactionOverrides: {
       thresholdPercent: bundle.resolvedAgent.config.compaction?.thresholdPercent,
     },
@@ -219,21 +219,34 @@ export async function turnStep(rawInput: TurnStepInput): Promise<DurableStepResu
 
   const adapterCtx = buildAdapterContext(adapter, ctx);
 
-  // Run the adapter's deliver hook for each queued payload and
-  // coalesce the resulting StepInput values.
+  // Run the adapter's deliver hook for each queued payload inside the same
+  // hydrated ALS scope as tools and event handlers. Durable channel policies
+  // can therefore use SessionContext, defineState, sandbox access, and other
+  // context-backed APIs before deciding whether the delivery starts a turn.
   let resolved: StepInput | undefined;
+  let deliverySessionChanged = false;
   if (input.input?.kind === "deliver") {
-    const results: StepInput[] = [];
-    for (const payload of input.input.payloads) {
-      const result = adapter.deliver
-        ? await adapter.deliver(payload, adapterCtx)
-        : defaultDeliverResult(payload);
-
-      if (result !== undefined && result !== null) {
-        results.push(result);
+    const delivery = input.input;
+    const beforeDeliverySession = initialSession;
+    const scoped = await withContextScope(ctx, initialSession, async (enrichedSession) => {
+      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<DurableStepResu
     const rekeyed = reconcileSessionContinuationToken(ctx, initialSession);
     const nextSerializedContext = serializeContext(ctx);
     const nextState =
-      rekeyed === initialSession
+      rekeyed === initialSession && !deliverySessionChanged
         ? input.sessionState
         : createDurableSessionState({ session: rekeyed });
 

packages/eve/src/public/channels/slack/index.ts

@@ -39,6 +39,13 @@ export {
   type SlackResetOptions,
   type SlackSessionTarget,
   type SlackThread,
+  type SlackStatefulThreadReplies,
+  type SlackStatefulThreadReplyContext,
+  type SlackStatefulThreadReplyResult,
+  type SlackStatelessThreadReplies,
+  type SlackThreadReplies,
+  type SlackThreadReplyContext,
+  type SlackThreadReplyResult,
   type SlackWebhookVerifier,
   type SlackWorkspaceHandle,
 } from "#public/channels/slack/slackChannel.js";

packages/eve/src/public/channels/slack/interactions.ts

@@ -326,7 +326,7 @@ function hasCardContent(block: Record<string, unknown>): boolean {
  * `onInteraction` callback for non-HITL clicks.
  */
 export interface InteractionHandlerDeps {
-  readonly config: SlackChannelConfig;
+  readonly config: SlackChannelConfig<unknown>;
 }
 
 /**