refactor(runtime): consolidate followup, gateway, and provider dedupe paths

2026-06-07 22:09:57 +00:00 · 2026-02-22 14:06:03 +00:00
parent 38752338dc
commit d116bcfb14
36 changed files with 848 additions and 908 deletions
--- a/src/slack/monitor/events/interactions.ts
+++ b/src/slack/monitor/events/interactions.ts
@@ -99,6 +99,11 @@ type SlackModalEventBase = {
 };

 type SlackModalInteractionKind = "view_submission" | "view_closed";
+type SlackModalEventHandlerArgs = { ack: () => Promise<void>; body: unknown };
+type RegisterSlackModalHandler = (
+  matcher: RegExp,
+  handler: (args: SlackModalEventHandlerArgs) => Promise<void>,
+) => void;

 function readOptionValues(options: unknown): string[] | undefined {
  if (!Array.isArray(options)) {
@@ -483,6 +488,24 @@ function emitSlackModalLifecycleEvent(params: {
  });
 }

+function registerModalLifecycleHandler(params: {
+  register: RegisterSlackModalHandler;
+  matcher: RegExp;
+  ctx: SlackMonitorContext;
+  interactionType: SlackModalInteractionKind;
+  contextPrefix: "slack:interaction:view" | "slack:interaction:view-closed";
+}) {
+  params.register(params.matcher, async ({ ack, body }: SlackModalEventHandlerArgs) => {
+    await ack();
+    emitSlackModalLifecycleEvent({
+      ctx: params.ctx,
+      body: body as SlackModalBody,
+      interactionType: params.interactionType,
+      contextPrefix: params.contextPrefix,
+    });
+  });
+}
+
 export function registerSlackInteractionEvents(params: { ctx: SlackMonitorContext }) {
  const { ctx } = params;
  if (typeof ctx.app.action !== "function") {
@@ -646,27 +669,20 @@ export function registerSlackInteractionEvents(params: { ctx: SlackMonitorContex
  if (typeof ctx.app.view !== "function") {
    return;
  }
+  const modalMatcher = new RegExp(`^${OPENCLAW_ACTION_PREFIX}`);

  // Handle OpenClaw modal submissions with callback_ids scoped by our prefix.
-  ctx.app.view(
-    new RegExp(`^${OPENCLAW_ACTION_PREFIX}`),
-    async ({ ack, body }: { ack: () => Promise<void>; body: unknown }) => {
-      await ack();
-      emitSlackModalLifecycleEvent({
-        ctx,
-        body: body as SlackModalBody,
-        interactionType: "view_submission",
-        contextPrefix: "slack:interaction:view",
-      });
-    },
-  );
+  registerModalLifecycleHandler({
+    register: (matcher, handler) => ctx.app.view(matcher, handler),
+    matcher: modalMatcher,
+    ctx,
+    interactionType: "view_submission",
+    contextPrefix: "slack:interaction:view",
+  });

  const viewClosed = (
    ctx.app as unknown as {
-      viewClosed?: (
-        matcher: RegExp,
-        handler: (args: { ack: () => Promise<void>; body: unknown }) => Promise<void>,
-      ) => void;
+      viewClosed?: RegisterSlackModalHandler;
    }
  ).viewClosed;
  if (typeof viewClosed !== "function") {
@@ -674,16 +690,11 @@ export function registerSlackInteractionEvents(params: { ctx: SlackMonitorContex
  }

  // Handle modal close events so agent workflows can react to cancelled forms.
-  viewClosed(
-    new RegExp(`^${OPENCLAW_ACTION_PREFIX}`),
-    async ({ ack, body }: { ack: () => Promise<void>; body: unknown }) => {
-      await ack();
-      emitSlackModalLifecycleEvent({
-        ctx,
-        body: body as SlackModalBody,
-        interactionType: "view_closed",
-        contextPrefix: "slack:interaction:view-closed",
-      });
-    },
-  );
+  registerModalLifecycleHandler({
+    register: viewClosed,
+    matcher: modalMatcher,
+    ctx,
+    interactionType: "view_closed",
+    contextPrefix: "slack:interaction:view-closed",
+  });
 }
--- a/src/slack/monitor/message-handler/dispatch.ts
+++ b/src/slack/monitor/message-handler/dispatch.ts
@@ -292,17 +292,7 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
        hasStreamedMessage = false;
      }

-      const replyThreadTs = replyPlan.nextThreadTs();
-      await deliverReplies({
-        replies: [payload],
-        target: prepared.replyTarget,
-        token: ctx.botToken,
-        accountId: account.accountId,
-        runtime,
-        textLimit: ctx.textLimit,
-        replyThreadTs,
-      });
-      replyPlan.markSent();
+      await deliverNormally(payload);
    },
    onError: (err, info) => {
      runtime.error?.(danger(`slack ${info.kind} reply failed: ${String(err)}`));
@@ -362,6 +352,18 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
    draftStream.update(trimmed);
    hasStreamedMessage = true;
  };
+  const onDraftBoundary =
+    useStreaming || !previewStreamingEnabled
+      ? undefined
+      : async () => {
+          if (hasStreamedMessage) {
+            draftStream.forceNewMessage();
+            hasStreamedMessage = false;
+            appendRenderedText = "";
+            appendSourceText = "";
+            statusUpdateCount = 0;
+          }
+        };

  const { queuedFinal, counts } = await dispatchInboundMessage({
    ctx: prepared.ctxPayload,
@@ -384,32 +386,8 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag
          : async (payload) => {
              updateDraftFromPartial(payload.text);
            },
-      onAssistantMessageStart: useStreaming
-        ? undefined
-        : !previewStreamingEnabled
-          ? undefined
-          : async () => {
-              if (hasStreamedMessage) {
-                draftStream.forceNewMessage();
-                hasStreamedMessage = false;
-                appendRenderedText = "";
-                appendSourceText = "";
-                statusUpdateCount = 0;
-              }
-            },
-      onReasoningEnd: useStreaming
-        ? undefined
-        : !previewStreamingEnabled
-          ? undefined
-          : async () => {
-              if (hasStreamedMessage) {
-                draftStream.forceNewMessage();
-                hasStreamedMessage = false;
-                appendRenderedText = "";
-                appendSourceText = "";
-                statusUpdateCount = 0;
-              }
-            },
+      onAssistantMessageStart: onDraftBoundary,
+      onReasoningEnd: onDraftBoundary,
    },
  });
  await draftStream.flush();
--- a/src/slack/monitor/message-handler/prepare.test.ts
+++ b/src/slack/monitor/message-handler/prepare.test.ts
@@ -168,6 +168,19 @@ describe("slack prepareSlackMessage inbound contract", () => {
    };
  }

+  function createThreadReplyMessage(overrides: Partial<SlackMessageEvent>): SlackMessageEvent {
+    return createSlackMessage({
+      channel: "C123",
+      channel_type: "channel",
+      thread_ts: "100.000",
+      ...overrides,
+    });
+  }
+
+  function prepareThreadMessage(ctx: SlackMonitorContext, overrides: Partial<SlackMessageEvent>) {
+    return prepareMessageWith(ctx, createThreadAccount(), createThreadReplyMessage(overrides));
+  }
+
  it("produces a finalized MsgContext", async () => {
    const message: SlackMessageEvent = {
      channel: "D123",
@@ -298,17 +311,10 @@ describe("slack prepareSlackMessage inbound contract", () => {
    });
    slackCtx.resolveChannelName = async () => ({ name: "general", type: "channel" });

-    const prepared = await prepareMessageWith(
-      slackCtx,
-      createThreadAccount(),
-      createSlackMessage({
-        channel: "C123",
-        channel_type: "channel",
-        text: "current message",
-        ts: "101.000",
-        thread_ts: "100.000",
-      }),
-    );
+    const prepared = await prepareThreadMessage(slackCtx, {
+      text: "current message",
+      ts: "101.000",
+    });

    expect(prepared).toBeTruthy();
    expect(prepared!.ctxPayload.IsFirstThreadTurn).toBe(true);
@@ -347,17 +353,11 @@ describe("slack prepareSlackMessage inbound contract", () => {
    slackCtx.resolveUserName = async () => ({ name: "Alice" });
    slackCtx.resolveChannelName = async () => ({ name: "general", type: "channel" });

-    const prepared = await prepareMessageWith(
-      slackCtx,
-      createThreadAccount(),
-      createSlackMessage({
-        channel: "C123",
-        channel_type: "channel",
-        text: "reply in old thread",
-        ts: "201.000",
-        thread_ts: "200.000",
-      }),
-    );
+    const prepared = await prepareThreadMessage(slackCtx, {
+      text: "reply in old thread",
+      ts: "201.000",
+      thread_ts: "200.000",
+    });

    expect(prepared).toBeTruthy();
    expect(prepared!.ctxPayload.IsFirstThreadTurn).toBeUndefined();
--- a/src/slack/monitor/slash.ts
+++ b/src/slack/monitor/slash.ts
@@ -147,6 +147,13 @@ function parseSlackCommandArgValue(raw?: string | null): {
  };
 }

+function buildSlackArgMenuOptions(choices: EncodedMenuChoice[]) {
+  return choices.map((choice) => ({
+    text: { type: "plain_text", text: choice.label.slice(0, 75) },
+    value: choice.value,
+  }));
+}
+
 function buildSlackCommandArgMenuBlocks(params: {
  title: string;
  command: string;
@@ -185,10 +192,7 @@ function buildSlackCommandArgMenuBlocks(params: {
              type: "overflow",
              action_id: SLACK_COMMAND_ARG_ACTION_ID,
              confirm: buildSlackArgMenuConfirm({ command: params.command, arg: params.arg }),
-              options: encodedChoices.map((choice) => ({
-                text: { type: "plain_text", text: choice.label.slice(0, 75) },
-                value: choice.value,
-              })),
+              options: buildSlackArgMenuOptions(encodedChoices),
            },
          ],
        },
@@ -238,10 +242,7 @@ function buildSlackCommandArgMenuBlocks(params: {
                    text:
                      index === 0 ? `Choose ${params.arg}` : `Choose ${params.arg} (${index + 1})`,
                  },
-                  options: choices.map((choice) => ({
-                    text: { type: "plain_text", text: choice.label.slice(0, 75) },
-                    value: choice.value,
-                  })),
+                  options: buildSlackArgMenuOptions(choices),
                },
              ],
            }),