From 7fc7b97a1c3459fdbcae9d52f30f4089fa32cace Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 20:53:44 +0000 Subject: [PATCH] slack: use assistant.threads.setStatus instead of a "Thinking..." message MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The agent posted a real ":hourglass_flowing_sand: Thinking..." message and mutated it via chat.update into the answer, the confirmation card, or an error. That placeholder notifies, shows up in thread history (and had to be filtered back out before feeding the model), and leaves a stale "Thinking..." artifact when a run dies mid-flight. Slack's assistant.threads.setStatus renders an ephemeral loading indicator instead. Since the March 2026 scope change it accepts chat:write, which the app already has, so it works in channels with no manifest or scope change. Dropping the placeholder means the reply is a fresh post rather than an update, so: - postThreadReply centralizes the cannot_reply_to_message top-level fallback, now shared by the answer, confirmation and error paths. - setStatus is best-effort — failures are logged and the run continues, since a cosmetic indicator must never silence the bot. A 60s refresh keeps it alive past Slack's ~2 minute drop, and it is cleared before the reply lands. - A second destructive draft in a thread now strips the old card's buttons and marks it superseded, rather than leaving two identical live cards sharing one action id. - store() takes a caller-supplied id so the card's blocks can be built before the message exists; replace() also updates the stored messageTs. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_018vZGKg3npYcKypYaYEDkJr --- .../src/libs/test/doubles/slack-test-state.ts | 2 + .../libs/test/doubles/slack-web-api.mock.ts | 9 + .../routes/slack/confirmation-store.test.ts | 45 +++-- .../src/routes/slack/confirmation-store.ts | 46 +++-- apps/server/src/routes/slack/handler.test.ts | 108 +++++++++-- apps/server/src/routes/slack/handler.ts | 180 +++++++++--------- apps/server/src/routes/slack/status.ts | 67 +++++++ 7 files changed, 317 insertions(+), 140 deletions(-) create mode 100644 apps/server/src/routes/slack/status.ts diff --git a/apps/server/src/libs/test/doubles/slack-test-state.ts b/apps/server/src/libs/test/doubles/slack-test-state.ts index 9d19743e..40b4e2b7 100644 --- a/apps/server/src/libs/test/doubles/slack-test-state.ts +++ b/apps/server/src/libs/test/doubles/slack-test-state.ts @@ -10,6 +10,7 @@ export interface SlackTestState { postMessageOverride: Override; updateOverride: Override; postEphemeralOverride: Override; + setStatusOverride: Override; runAgentOverride: (() => Promise) | null; repliesImpl: () => Promise; } @@ -22,6 +23,7 @@ if (!g.__slackTestState) { postMessageOverride: null, updateOverride: null, postEphemeralOverride: null, + setStatusOverride: null, runAgentOverride: null, repliesImpl: () => Promise.resolve({ diff --git a/apps/server/src/libs/test/doubles/slack-web-api.mock.ts b/apps/server/src/libs/test/doubles/slack-web-api.mock.ts index c4479e9e..61c8a94a 100644 --- a/apps/server/src/libs/test/doubles/slack-web-api.mock.ts +++ b/apps/server/src/libs/test/doubles/slack-web-api.mock.ts @@ -20,6 +20,15 @@ export class WebClient { return Promise.resolve(); }, }; + assistant = { + threads: { + setStatus: (args: Record) => { + if (s.setStatusOverride) return s.setStatusOverride(args); + s.calls.push({ method: "setStatus", args }); + return Promise.resolve({ ok: true }); + }, + }, + }; conversations = { replies: () => s.repliesImpl(), }; diff --git a/apps/server/src/routes/slack/confirmation-store.test.ts b/apps/server/src/routes/slack/confirmation-store.test.ts index 93e7a211..a8d32f0d 100644 --- a/apps/server/src/routes/slack/confirmation-store.test.ts +++ b/apps/server/src/routes/slack/confirmation-store.test.ts @@ -5,6 +5,7 @@ import { consume, findByThread, get, + newActionId, replace, store, } from "./confirmation-store"; @@ -13,8 +14,9 @@ import type { PendingAction } from "./confirmation-store"; const redisStore = (globalThis as Record) .__testRedisStore as Map; -function makePendingInput(): Omit { +function makePendingInput(): Omit { return { + id: newActionId(), workspaceId: 1, botToken: "xoxb-test-token", channelId: "C123", @@ -39,10 +41,10 @@ describe("confirmation-store", () => { }); describe("store", () => { - test("returns an action id", async () => { - const id = await store(makePendingInput()); - expect(typeof id).toBe("string"); - expect(id.length).toBeGreaterThan(0); + test("returns the caller-supplied action id", async () => { + const input = makePendingInput(); + const id = await store(input); + expect(id).toBe(input.id); }); test("saves action and thread index to redis", async () => { @@ -143,29 +145,38 @@ describe("confirmation-store", () => { }); describe("replace", () => { - test("replaces the payload on an existing pending", async () => { + test("replaces the payload and message ts on an existing pending", async () => { const input = makePendingInput(); const id = await store(input); - await replace(id, { - toolName: "add_status_report_update", - input: { - statusReportId: 42, - status: "identified", - message: "Root cause found", + await replace( + id, + { + toolName: "add_status_report_update", + input: { + statusReportId: 42, + status: "identified", + message: "Root cause found", + }, }, - }); + "9999999999.000001", + ); const result = await consume(id); expect(result).toBeDefined(); expect(result?.payload.toolName).toBe("add_status_report_update"); + expect(result?.messageTs).toBe("9999999999.000001"); }); test("does nothing for unknown id", async () => { - await replace("nonexistent", { - toolName: "resolve_status_report", - input: { statusReportId: 1, message: "fixed" }, - }); + await replace( + "nonexistent", + { + toolName: "resolve_status_report", + input: { statusReportId: 1, message: "fixed" }, + }, + "1.1", + ); expect(redisStore.size).toBe(0); }); }); diff --git a/apps/server/src/routes/slack/confirmation-store.ts b/apps/server/src/routes/slack/confirmation-store.ts index 1f17f7c7..309c7d02 100644 --- a/apps/server/src/routes/slack/confirmation-store.ts +++ b/apps/server/src/routes/slack/confirmation-store.ts @@ -31,12 +31,21 @@ export type PendingAction = z.infer; * adapter can be exercised without `@/libs/clients`. */ export interface CarrierStore { - put(action: Omit): Promise; + /** `id` is caller-supplied so the card's blocks can be built before it exists. */ + put(action: Omit): Promise; get(id: string): Promise; /** Atomic getdel — defends against double-click double-execution. */ consume(id: string): Promise; findByThread(threadTs: string): Promise; - replace(id: string, payload: PendingPayload): Promise; + replace( + id: string, + payload: PendingPayload, + messageTs: string, + ): Promise; +} + +export function newActionId(): string { + return nanoid(); } const TTL_SECONDS = 5 * 60; @@ -56,19 +65,18 @@ function parse(raw: unknown): PendingAction | undefined { export function createRedisCarrierStore(): CarrierStore { return { async put(action) { - const id = nanoid(); - const pending: PendingAction = { ...action, id, createdAt: Date.now() }; + const pending: PendingAction = { ...action, createdAt: Date.now() }; await Promise.all([ - redis.set(`${ACTION_PREFIX}${id}`, JSON.stringify(pending), { + redis.set(`${ACTION_PREFIX}${action.id}`, JSON.stringify(pending), { ex: TTL_SECONDS, }), - redis.set(`${THREAD_PREFIX}${action.threadTs}`, id, { + redis.set(`${THREAD_PREFIX}${action.threadTs}`, action.id, { ex: TTL_SECONDS, }), ]); - return id; + return action.id; }, async get(id) { @@ -103,7 +111,7 @@ export function createRedisCarrierStore(): CarrierStore { return parse(raw); }, - async replace(id, payload) { + async replace(id, payload, messageTs) { const raw = await redis.get(`${ACTION_PREFIX}${id}`); if (!raw) return; @@ -111,6 +119,7 @@ export function createRedisCarrierStore(): CarrierStore { if (!existing) return; existing.payload = payload; + existing.messageTs = messageTs; existing.createdAt = Date.now(); await Promise.all([ @@ -133,11 +142,10 @@ export function createMemoryCarrierStore(): CarrierStore { return { async put(action) { - const id = nanoid(); - const pending: PendingAction = { ...action, id, createdAt: Date.now() }; - actions.set(id, pending); - threads.set(action.threadTs, id); - return id; + const pending: PendingAction = { ...action, createdAt: Date.now() }; + actions.set(action.id, pending); + threads.set(action.threadTs, action.id); + return action.id; }, async get(id) { @@ -163,12 +171,13 @@ export function createMemoryCarrierStore(): CarrierStore { return action; }, - async replace(id, payload) { + async replace(id, payload, messageTs) { const existing = actions.get(id); if (!existing) return; actions.set(id, { ...existing, payload, + messageTs, createdAt: Date.now(), }); }, @@ -180,7 +189,7 @@ export function createMemoryCarrierStore(): CarrierStore { const defaultStore = createRedisCarrierStore(); export const store = ( - action: Omit, + action: Omit, ): Promise => defaultStore.put(action); export const get = (id: string): Promise => defaultStore.get(id); @@ -189,5 +198,8 @@ export const consume = (id: string): Promise => export const findByThread = ( threadTs: string, ): Promise => defaultStore.findByThread(threadTs); -export const replace = (id: string, payload: PendingPayload): Promise => - defaultStore.replace(id, payload); +export const replace = ( + id: string, + payload: PendingPayload, + messageTs: string, +): Promise => defaultStore.replace(id, payload, messageTs); diff --git a/apps/server/src/routes/slack/handler.test.ts b/apps/server/src/routes/slack/handler.test.ts index 5c6d4635..af44d067 100644 --- a/apps/server/src/routes/slack/handler.test.ts +++ b/apps/server/src/routes/slack/handler.test.ts @@ -51,6 +51,7 @@ describe("handleSlackEvent", () => { slackTestState.calls = []; slackTestState.postMessageOverride = null; slackTestState.updateOverride = null; + slackTestState.setStatusOverride = null; slackTestState.runAgentOverride = null; slackTestState.resolveWorkspace = (teamId: string) => { if (teamId === "T_KNOWN") { @@ -177,10 +178,16 @@ describe("handleSlackEvent", () => { }); await new Promise((r) => setTimeout(r, 100)); - const thinking = slackTestState.calls.filter( + const replies = slackTestState.calls.filter( (m) => m.method === "postMessage", ); - expect(thinking.length).toBe(1); + expect(replies.length).toBe(1); + + // The loading indicator is set once and cleared once. + const statuses = slackTestState.calls.filter( + (m) => m.method === "setStatus", + ); + expect(statuses.length).toBe(2); }); test("handles app_uninstalled event", async () => { @@ -471,7 +478,7 @@ describe("handleSlackEvent", () => { expect(fallbackPost).toBeDefined(); }); - test("returns early on non-recoverable postMessage error", async () => { + test("does not throw on a non-recoverable postMessage error", async () => { slackTestState.postMessageOverride = () => { const err = new Error("An API error occurred: channel_not_found"); Object.assign(err, { @@ -497,10 +504,89 @@ describe("handleSlackEvent", () => { expect(res.status).toBe(200); await new Promise((r) => setTimeout(r, 100)); - const updateMessages = slackTestState.calls.filter( - (m) => m.method === "update", + // channel_not_found isn't cannot_reply_to_message, so no top-level retry. + const posts = slackTestState.calls.filter( + (m) => m.method === "postMessage", ); - expect(updateMessages.length).toBe(0); + expect(posts.length).toBe(0); + }); + + test("sets and clears the thinking status around the agent run", async () => { + const res = await signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: `evt_status_${Date.now()}`, + event: { + type: "app_mention", + text: "<@UBOT> hello", + user: "U1", + channel: "C1", + ts: "500.10", + }, + }); + + expect(res.status).toBe(200); + await new Promise((r) => setTimeout(r, 100)); + + const statuses = slackTestState.calls.filter( + (m) => m.method === "setStatus", + ); + expect(statuses.length).toBe(2); + expect(statuses[0].args.channel_id).toBe("C1"); + expect(statuses[0].args.thread_ts).toBe("500.10"); + expect(statuses[0].args.status).toBe("is thinking..."); + expect( + (statuses[0].args.loading_messages as string[]).length, + ).toBeGreaterThan(0); + expect(statuses[1].args.status).toBe(""); + + // Cleared before the reply lands, so a refresh tick can't resurrect it. + const clearIndex = slackTestState.calls.findIndex( + (m) => m.method === "setStatus" && m.args.status === "", + ); + const postIndex = slackTestState.calls.findIndex( + (m) => m.method === "postMessage", + ); + expect(clearIndex).toBeLessThan(postIndex); + + // No placeholder message — one post, carrying the answer. + const posts = slackTestState.calls.filter( + (m) => m.method === "postMessage", + ); + expect(posts.length).toBe(1); + expect(posts[0].args.thread_ts).toBe("500.10"); + }); + + test("still replies when setStatus fails", async () => { + slackTestState.setStatusOverride = () => { + const err = new Error("An API error occurred: channel_not_found"); + Object.assign(err, { + code: "slack_webapi_platform_error", + data: { ok: false, error: "channel_not_found" }, + }); + return Promise.reject(err); + }; + + const res = await signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: `evt_statusfail_${Date.now()}`, + event: { + type: "app_mention", + text: "<@UBOT> hello", + user: "U1", + channel: "C1", + ts: `${Date.now()}.22`, + }, + }); + + expect(res.status).toBe(200); + await new Promise((r) => setTimeout(r, 100)); + + const posts = slackTestState.calls.filter( + (m) => m.method === "postMessage", + ); + expect(posts.length).toBe(1); }); test("shows error message when runAgent throws", async () => { @@ -523,19 +609,19 @@ describe("handleSlackEvent", () => { expect(res.status).toBe(200); await new Promise((r) => setTimeout(r, 100)); - const errorUpdate = slackTestState.calls.find( + const errorPost = slackTestState.calls.find( (m) => - m.method === "update" && + m.method === "postMessage" && typeof m.args.text === "string" && m.args.text.includes("Something went wrong"), ); - expect(errorUpdate).toBeDefined(); + expect(errorPost).toBeDefined(); }); - test("does not throw when both runAgent and error update fail", async () => { + test("does not throw when both runAgent and the error post fail", async () => { slackTestState.runAgentOverride = () => Promise.reject(new Error("agent exploded")); - slackTestState.updateOverride = () => { + slackTestState.postMessageOverride = () => { const err = new Error("An API error occurred: channel_not_found"); Object.assign(err, { code: "slack_webapi_platform_error", diff --git a/apps/server/src/routes/slack/handler.ts b/apps/server/src/routes/slack/handler.ts index 8d6a54da..ffeb3927 100644 --- a/apps/server/src/routes/slack/handler.ts +++ b/apps/server/src/routes/slack/handler.ts @@ -7,15 +7,22 @@ import { z } from "zod"; import { runAgent } from "./agent"; import { + type Block, buildConfirmationBlocks, getConfirmationText, type RefResolvers, } from "./blocks"; -import { findByThread, replace, store } from "./confirmation-store"; +import { + findByThread, + newActionId, + replace, + store, +} from "./confirmation-store"; import type { PendingPayload } from "./confirmation-store"; import { publishHomeView } from "./home"; import { getComponentNames, getPageDashboardLink } from "./page-urls"; import { getRegistryTool, isSlackToolDraft } from "./registry-runner"; +import { startThinkingStatus } from "./status"; import { resolveWorkspace } from "./workspace-resolver"; function makeRefResolvers(workspaceId: number): RefResolvers { @@ -83,6 +90,46 @@ function isSlackPlatformError(err: unknown, errorCode: string): boolean { return parsed.success && parsed.data.data.error === errorCode; } +/** + * Posts into the thread, falling back to a top-level message when Slack + * refuses the reply. Returns the new message ts, or undefined if both + * attempts failed. + */ +async function postThreadReply( + slack: WebClient, + channel: string, + threadTs: string, + message: { text: string; blocks?: Block[] }, +): Promise { + try { + const posted = await slack.chat.postMessage({ + channel, + thread_ts: threadTs, + ...message, + }); + return posted.ts; + } catch (err) { + if (!isSlackPlatformError(err, "cannot_reply_to_message")) { + logger.error("slack failed to post reply", { error: err, channel }); + return undefined; + } + logger.warn("slack cannot reply to message, falling back to top-level", { + channel, + threadTs, + }); + try { + const fallback = await slack.chat.postMessage({ channel, ...message }); + return fallback.ts; + } catch (fallbackErr) { + logger.error("slack failed to post fallback reply", { + error: fallbackErr, + channel, + }); + return undefined; + } + } +} + export async function handleSlackEvent(c: Context) { const body = c.get("slackBody") as SlackEvent; @@ -198,53 +245,7 @@ async function processEvent(body: SlackEvent) { user: event.user, }); - let thinkingTs: string | undefined; - try { - const thinkingMsg = await slack.chat.postMessage({ - channel: event.channel, - thread_ts: threadTs, - text: ":hourglass_flowing_sand: Thinking...", - }); - thinkingTs = thinkingMsg.ts; - } catch (err) { - if (isSlackPlatformError(err, "cannot_reply_to_message")) { - logger.warn("slack cannot reply to message, falling back to top-level", { - channel: event.channel, - teamId, - threadTs, - }); - try { - const fallbackMsg = await slack.chat.postMessage({ - channel: event.channel, - text: ":hourglass_flowing_sand: Thinking...", - }); - thinkingTs = fallbackMsg.ts; - } catch (fallbackErr) { - logger.error("slack failed to post fallback thinking message", { - error: fallbackErr, - channel: event.channel, - teamId, - }); - return; - } - } else { - logger.error("slack failed to post thinking message", { - error: err, - channel: event.channel, - teamId, - threadTs, - }); - return; - } - } - - if (!thinkingTs) { - logger.error("slack thinking message returned no ts", { - channel: event.channel, - teamId, - }); - return; - } + const status = await startThinkingStatus(slack, event.channel, threadTs); try { let thread: ThreadMessage[] = []; @@ -254,9 +255,7 @@ async function processEvent(body: SlackEvent) { ts: event.thread_ts, limit: 100, }); - thread = ((replies.messages ?? []) as ThreadMessage[]).filter( - (msg) => msg.ts !== thinkingTs, - ); + thread = (replies.messages ?? []) as ThreadMessage[]; } else { thread = [{ user: event.user, text: event.text, ts: event.ts }]; } @@ -292,6 +291,9 @@ async function processEvent(body: SlackEvent) { isSlackToolDraft(tr.result), ); + // Clear the indicator before posting so a refresh tick can't resurrect it. + await status.stop(); + if (confirmationResult) { logger.info("slack confirmation requested", { teamId, @@ -303,16 +305,13 @@ async function processEvent(body: SlackEvent) { slack, event.channel, threadTs, - thinkingTs, event.user ?? "", resolved.workspace.id, resolved.botToken, confirmationResult, ); } else { - await slack.chat.update({ - channel: event.channel, - ts: thinkingTs, + await postThreadReply(slack, event.channel, threadTs, { text: result.text || "Done!", }); logger.info("slack response sent", { @@ -328,21 +327,10 @@ async function processEvent(body: SlackEvent) { teamId, threadTs, }); - if (thinkingTs) { - await slack.chat - .update({ - channel: event.channel, - ts: thinkingTs, - text: ":x: Something went wrong. Please try again.", - }) - .catch((updateErr: unknown) => { - logger.error("slack failed to update error message", { - error: updateErr, - channel: event.channel, - thinkingTs, - }); - }); - } + await status.stop(); + await postThreadReply(slack, event.channel, threadTs, { + text: ":x: Something went wrong. Please try again.", + }); } } @@ -350,7 +338,6 @@ async function handleConfirmation( slack: WebClient, channel: string, threadTs: string, - thinkingTs: string, userId: string, workspaceId: number, botToken: string, @@ -363,9 +350,7 @@ async function handleConfirmation( logger.error("slack: registry tool not found", { toolName: draft.toolName, }); - await slack.chat.update({ - channel, - ts: thinkingTs, + await postThreadReply(slack, channel, threadTs, { text: ":x: Something went wrong. Please try again.", }); return; @@ -384,39 +369,44 @@ async function handleConfirmation( // event throttling. Cross-process dedup is *not* covered; see note in // processedEvents. const existing = await findByThread(threadTs); - if (existing) { - await replace(existing.id, payload); + const actionId = existing?.id ?? newActionId(); - const blocks = await buildConfirmationBlocks({ - actionId: existing.id, - tool, - input: draft.displayInput, - resolvers: makeRefResolvers(workspaceId), - }); - await slack.chat.update({ channel, ts: thinkingTs, text, blocks }); + const blocks = await buildConfirmationBlocks({ + actionId, + tool, + input: draft.displayInput, + resolvers: makeRefResolvers(workspaceId), + }); + + // Strip the old card's buttons first: both cards carry the same actionId, + // so leaving it live would let one click consume the other's action. + if (existing) { await slack.chat.update({ channel, ts: existing.messageTs, - text, - blocks, + text: ":no_entry_sign: Superseded by a newer request.", + blocks: [], }); + } + + const messageTs = await postThreadReply(slack, channel, threadTs, { + text, + blocks, + }); + if (!messageTs) return; + + if (existing) { + await replace(existing.id, payload, messageTs); } else { - const actionId = await store({ + await store({ + id: actionId, workspaceId, botToken, channelId: channel, threadTs, - messageTs: thinkingTs, + messageTs, userId, payload, }); - - const blocks = await buildConfirmationBlocks({ - actionId, - tool, - input: draft.displayInput, - resolvers: makeRefResolvers(workspaceId), - }); - await slack.chat.update({ channel, ts: thinkingTs, text, blocks }); } } diff --git a/apps/server/src/routes/slack/status.ts b/apps/server/src/routes/slack/status.ts new file mode 100644 index 00000000..e129004f --- /dev/null +++ b/apps/server/src/routes/slack/status.ts @@ -0,0 +1,67 @@ +import { getLogger } from "@logtape/logtape"; +import type { WebClient } from "@slack/web-api"; + +const logger = getLogger("api-server"); + +// Slack drops the status ~2 minutes after the last setStatus call if no +// message has been posted, so refresh it while the agent is still running. +const REFRESH_MS = 60_000; + +const STATUS = "is thinking..."; + +const LOADING_MESSAGES = [ + "Reading the thread...", + "Checking your status pages...", + "Drafting an update...", +]; + +export interface ThinkingStatus { + /** Idempotent — clears the refresh timer and removes the indicator. */ + stop(): Promise; +} + +/** + * Shows Slack's native loading indicator on a thread for the duration of an + * agent run. Purely cosmetic: every call swallows its error so a setStatus + * outage can never stop the bot from replying. + */ +export async function startThinkingStatus( + slack: WebClient, + channelId: string, + threadTs: string, +): Promise { + const setStatus = (status: string, loadingMessages?: string[]) => + slack.assistant.threads + .setStatus({ + channel_id: channelId, + thread_ts: threadTs, + status, + ...(loadingMessages ? { loading_messages: loadingMessages } : {}), + }) + .then(() => undefined) + .catch((error: unknown) => { + logger.warn("slack failed to set thinking status", { + error, + channel: channelId, + threadTs, + }); + }); + + await setStatus(STATUS, LOADING_MESSAGES); + + let stopped = false; + const timer = setInterval(() => { + void setStatus(STATUS, LOADING_MESSAGES); + }, REFRESH_MS); + + return { + async stop() { + if (stopped) return; + stopped = true; + clearInterval(timer); + // Posting in-thread clears the indicator on its own, but the + // top-level fallback in the handler posts outside the thread. + await setStatus(""); + }, + }; +} -- 2.51.2