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(""); + }, + }; +}