Something went wrong. Try again.
[READ-ONLY] Mirror of https://github.com/openstatusHQ/openstatus. 🫖 Status page with uptime monitoring & API monitoring as code 🫖 openstatus.dev
bun drizzle-orm monitoring monitoring-as-code nextjs observability on-call open-source shadcn-ui status-page statuspage synthetic-monitoring tinybird turso uptime uptime-checker uptime-monitor
Something went wrong. Try again.
TypeScript
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523import { getLogger } from "@logtape/logtape";import { getIncidentBySlackChannel } from "@openstatus/services/incident";import { missingSlackScopes, uninstallSlackTeam,} from "@openstatus/services/integration";import { WebClient } from "@slack/web-api";import type { Context } from "hono";import { z } from "zod";
import { runInBackground } from "@/libs/background";import { redis } from "@/libs/clients";
import { type AgentEvents, runAgent } from "./agent";import { greetOnce, setAssistantStatus, setSessionStatus } from "./assistant";import { type Block, buildAnswerMessage, buildConfirmationBlocks, buildLinkAccountBlocks, getConfirmationText, LINK_ACCOUNT_TEXT, type RefResolvers,} from "./blocks";import { channelContextTooling, channelHistoryNote, channelHistoryTooling, contextChannelId, contextUserId, fetchChannelHistory, forgetContext, recallContext, rememberContext,} from "./channel-context";import type { SlackConfig, SlackEnv } from "./config";import { makeRefResolvers } from "./confirmation-card";import { draftKey, findByThread, listCreated, replace, store,} from "./confirmation-store";import type { PendingPayload } from "./confirmation-store";import { homeIncidents, publishHomeView, publishLinkAccountView } from "./home";import { handleChannelGone, handlePinReaction } from "./incident-events";import { getComponentNames, getPageDashboardLink, getStatusReportLink,} from "./page-urls";import { getRegistryTool, isSlackToolDraft, type SlackToolDraft,} from "./registry-runner";import { claimLinkCardWindow, claimPlanNoticeWindow, linkAccountUrl, planRequiredMessage, releaseLinkCardWindow, requireSlackMember, type SlackActor, slackAgentAllowed,} from "./require-slack-member";import { abortTurn, broadcastStop, endTurn, startTurn } from "./running-turns";import { permalinkFor } from "./service-adapter";import { linkedMessagesNote, slackLinkTools } from "./slack-links";import { createdNote } from "./thread-memory";import { buildThreadTitle, isThreadTitled, markThreadTitled, renameThread,} from "./thread-title";import { buildWelcomeBlocks, CATCH_UP_REQUEST, WELCOME_TEXT } from "./welcome";import { resolveWorkspace, type SlackWorkspace } from "./workspace-resolver";
const logger = getLogger("api-server");
const DEDUP_TTL_SECONDS = 10 * 60;
/** * Whether this event was already claimed, by this instance or another. If * Redis is unreachable the event is processed: a duplicate turn beats a * dropped one. */async function isDuplicate(key: string): Promise<boolean> { try { const claimed = await redis.set(`slack:event:${key}`, "1", { nx: true, ex: DEDUP_TTL_SECONDS, }); return claimed === null; } catch (error) { logger.warn("slack dedup unavailable", { error, key }); return false; }}
const slackEventSchema = z.object({ type: z.string(), event: z .object({ type: z.string(), subtype: z.string().optional(), text: z.string().optional(), user: z.string().optional(), // `member_joined_channel`: absent when the member joined on their own. inviter: z.string().optional(), channel: z.string().optional(), channel_type: z.string().optional(), ts: z.string().optional(), thread_ts: z.string().optional(), bot_id: z.string().optional(), tab: z.string().optional(), reaction: z.string().optional(), item: z .object({ type: z.string(), channel: z.string().optional(), ts: z.string().optional(), }) .optional(), // `tokens_revoked`: which tokens went. Only a revoked bot token matters. tokens: z .object({ bot: z.array(z.string()).optional(), oauth: z.array(z.string()).optional(), }) .optional(), assistant_thread: z .object({ channel_id: z.string(), thread_ts: z.string(), user_id: z.string().optional(), }) .optional(), // `app_context_changed` — what the user has on screen. Absent entities // (`"context": {}`) mean they moved somewhere with nothing to track. context: z .object({ entities: z .array( z.object({ type: z.string().optional(), value: z.string().optional(), }), ) .optional(), }) .optional(), }) .optional(), event_id: z.string().optional(), team_id: z.string().optional(), challenge: z.string().optional(), // `app_context_changed` carries no `event.user`; the human is in here. authorizations: z .array( z.object({ user_id: z.string().optional(), is_bot: z.boolean().optional(), }), ) .optional(),});
type SlackEvent = z.infer<typeof slackEventSchema>;
const threadMessageSchema = z.object({ user: z.string().optional(), bot_id: z.string().optional(), text: z.string().optional(), ts: z.string().optional(),});
type ThreadMessage = z.infer<typeof threadMessageSchema>;
const slackPlatformErrorSchema = z.object({ code: z.literal("slack_webapi_platform_error"), data: z.object({ error: z.string(), }),});
function isSlackPlatformError(err: unknown, errorCode: string): boolean { const parsed = slackPlatformErrorSchema.safeParse(err); return parsed.success && parsed.data.data.error === errorCode;}
// Asking permission to run a write tool ("shall I go ahead and publish this?").const PERMISSION_QUESTION = /\b(shall i|should i|do you want me to|would you like me to|want me to|ready for me to|can i)\b[^?]*\b(go ahead|proceed|publish|create|post|send|schedule|submit|add|resolve|update)\b/i;
// A change written out as message text rather than passed to a tool.const PROSE_DRAFT_FIELDS = /\*{0,2}(title|status|message|impact|components?|from|to)\*{0,2}\s*:/gi;
/** * Heuristic for the "model drafted in prose instead of calling the tool" failure * — logging only, so false positives are cheap. Both signals are required: a * plain answer can end in a question, and a list of reports can look tabular. */export function looksLikeUncardedDraft(text: string): boolean { if (!text) return false; if (!PERMISSION_QUESTION.test(text)) return false; return (text.match(PROSE_DRAFT_FIELDS) ?? []).length >= 2;}
const HISTORY_BEFORE_MENTION = 50;const HISTORY_BEFORE_THREAD = 20;
// Bounds the Slack calls (and the agent's context) on very long threads.const MAX_THREAD_PAGES = 5;
// Replies come oldest first, so a single page would miss the latest messages// of a long thread — the ones the agent is being asked about.async function fetchThread( slack: WebClient, channel: string, threadTs: string,): Promise<ThreadMessage[]> { const messages: ThreadMessage[] = []; let cursor: string | undefined; for (let page = 0; page < MAX_THREAD_PAGES; page++) { const replies = await slack.conversations.replies({ channel, ts: threadTs, limit: 100, cursor, }); messages.push(...((replies.messages ?? []) as ThreadMessage[])); cursor = replies.response_metadata?.next_cursor || undefined; if (!replies.has_more || !cursor) break; } return messages;}
/** * Whether an untagged channel-thread message is the user answering the agent * (e.g. "API" after "Which status page — API or Marketing?"). True only when * the agent posted the message right before it and its author is whoever first * mentioned the agent in the thread. Anything else still needs a mention, so * the agent stays out of the humans' side of an incident thread. */export function isAnswerToAgent( thread: ThreadMessage[], message: { ts: string; user?: string }, botUserId: string,): boolean { if (!message.user || !botUserId) return false; const index = thread.findIndex((m) => m.ts === message.ts); const earlier = index === -1 ? thread : thread.slice(0, index);
const previous = earlier.at(-1); if (previous?.user !== botUserId) return false;
const starter = earlier.find( (m) => m.user !== botUserId && m.text?.includes(`<@${botUserId}>`), ); return starter?.user === message.user;}
export async function handleSlackEvent(c: Context<SlackEnv>) { const body = c.get("slackBody") as SlackEvent; const config = c.get("slackConfig");
if (body.type === "url_verification") { return c.json({ challenge: body.challenge }); }
if (body.type !== "event_callback") { return c.json({ ok: true }); }
if (body.event_id && (await isDuplicate(body.event_id))) { return c.json({ ok: true }); }
runInBackground("event", () => processEvent(body, config), { teamId: body.team_id, eventId: body.event_id, });
return c.json({ ok: true });}
/** Tells the agent which incident a channel belongs to, when it has one. */async function boundIncidentNote(args: { workspace: SlackWorkspace["workspace"]; actor: SlackActor; teamId: string; channelId: string | undefined;}): Promise<string | undefined> { if (!args.channelId) return undefined; const bound = await getIncidentBySlackChannel({ ctx: { workspace: args.workspace, actor: args.actor }, input: { teamId: args.teamId, channelId: args.channelId }, }).catch(() => undefined); if (!bound) return undefined; const report = bound.statusReport ? ` Its status report is "${bound.statusReport.title}" (id ${bound.statusReport.id}, ${bound.statusReport.status}).` : " It has no status report yet."; return `Incident channel: <#${args.channelId}> belongs to managed incident "${bound.title}" (id ${bound.id}, ${bound.severity}, ${bound.status}${bound.closedAt ? ", closed" : ""}).${report} Notes and status changes discussed here are about this incident: use id ${bound.id} without asking.`;}
/** * Tells an unlinked Slack user how to link their account. Passive surfaces * (mentions, DMs) send it at most once per window, so a chatty user isn't * flooded with cards. */async function sendLinkCard(args: { config: SlackConfig; workspaceId: number; teamId: string; slackUserId: string; post: (message: { text: string; blocks: Block[]; }) => Promise<{ ok?: boolean }>;}): Promise<void> { const { config, workspaceId, teamId, slackUserId, post } = args; if (!(await claimLinkCardWindow(teamId, slackUserId))) return; try { const url = await linkAccountUrl(config, { workspaceId, teamId, slackUserId, }); await post({ text: LINK_ACCOUNT_TEXT, blocks: buildLinkAccountBlocks(url), }); } catch (err) { // Otherwise a transient failure silences the card for the whole window. await releaseLinkCardWindow(teamId, slackUserId).catch(() => undefined); throw err; } logger.info("slack link card sent", { teamId, slackUserId });}
/** * The welcome card's "Catch up": a visible message in the channel, then a turn * in its thread as if the clicker had mentioned the agent there. The channel * is only read because someone asked. */export async function catchUpChannel(args: { teamId: string; channelId: string; slackUserId: string; config: SlackConfig;}): Promise<void> { const { teamId, channelId, slackUserId, config } = args; const resolved = await resolveWorkspace(teamId); if (!resolved) return; const slack = new WebClient(resolved.botToken); // The welcome card is already gone: say why nothing happens. if (!slackAgentAllowed(resolved.workspace)) { await slack.chat.postEphemeral({ channel: channelId, user: slackUserId, ...planRequiredMessage(config), }); return; } const actor = await requireSlackMember({ workspace: resolved.workspace, teamId, slackUserId, slack, }); if (!actor) { await sendLinkCard({ config, workspaceId: resolved.workspace.id, teamId, slackUserId, post: (message) => slack.chat.postEphemeral({ channel: channelId, user: slackUserId, ...message, }), }); return; } const root = await slack.chat.postMessage({ channel: channelId, text: `:eyes: Catching up on <#${channelId}> for <@${slackUserId}>.`, }); if (!root.ts) return; await processEvent( { type: "event_callback", team_id: teamId, event: { type: "app_mention", user: slackUserId, channel: channelId, channel_type: "channel", ts: root.ts, text: `<@${resolved.botUserId}> ${CATCH_UP_REQUEST}`, }, }, config, );}
async function processEvent(body: SlackEvent, config: SlackConfig) { const event = body.event; if (!event) return;
if (event.type === "app_uninstalled" || event.type === "tokens_revoked") { const teamId = body.team_id; if (!teamId) return; if (event.type === "tokens_revoked" && !event.tokens?.bot?.length) return; const result = await uninstallSlackTeam({ input: { teamId }, job: `slack-${event.type}`, }); logger.info("slack integration cleaned up", { teamId, ...result }); return; }
if (event.type === "reaction_added") { const teamId = body.team_id; const { item, user, reaction } = event; if (!teamId || !user || !reaction || item?.type !== "message") return; if (!item.channel || !item.ts) return; const resolved = await resolveWorkspace(teamId); if (!resolved) return; await handlePinReaction({ resolved, config, teamId, slackUserId: user, reaction, channel: item.channel, ts: item.ts, }); return; }
if (event.type === "channel_archive" || event.type === "channel_deleted") { const teamId = body.team_id; if (!teamId || !event.channel) return; const resolved = await resolveWorkspace(teamId); if (!resolved) return; await handleChannelGone({ resolved, teamId, channel: event.channel, slackUserId: event.user, }); return; }
// The bot also joins channels on its own (incident channels, `subscribe`); // those carry no inviter, so only a person adding it gets the welcome. if (event.type === "member_joined_channel") { const teamId = body.team_id; const { channel, inviter } = event; if (!teamId || !channel || !inviter) return; const resolved = await resolveWorkspace(teamId); if (!resolved || event.user !== resolved.botUserId) return; if (inviter === resolved.botUserId) return; await new WebClient(resolved.botToken).chat.postEphemeral({ channel, user: inviter, text: WELCOME_TEXT, blocks: buildWelcomeBlocks(resolved.botUserId), }); logger.info("slack welcome sent", { teamId, channel }); return; }
// Both tabs of the app's DM arrive here. The agent experience has no // "thread started" event, so opening the Messages tab is where a first-time // user gets greeted; `greetOnce` makes the repeat opens harmless. if (event.type === "app_home_opened") { const tab = event.tab ?? "home"; if (tab !== "home" && tab !== "messages") return; const teamId = body.team_id; const userId = event.user; if (!teamId || !userId) return; const resolved = await resolveWorkspace(teamId); if (!resolved) return; const slack = new WebClient(resolved.botToken); const member = { workspace: resolved.workspace, teamId, slack };
if (tab === "messages") { const channel = event.channel; if (!channel) return; const actor = await requireSlackMember({ ...member, slackUserId: userId, }); if (!actor) { await sendLinkCard({ config, workspaceId: resolved.workspace.id, teamId, slackUserId: userId, post: (message) => slack.chat.postMessage({ channel, ...message }), }).catch((error) => logger.error("slack failed to send link card", { error, teamId }), ); return; } try { await greetOnce({ slack, teamId, userId, channel, }); } catch (err) { logger.error("slack failed to greet user", { error: err, teamId }); } return; }
try { const actor = await requireSlackMember({ ...member, slackUserId: userId, }); if (actor) { const needsReconnect = missingSlackScopes(resolved.scopes).length > 0; await publishHomeView(slack, userId, { reconnectUrl: needsReconnect ? `${config.dashboardUrl}/settings/integrations` : undefined, openIncidents: await homeIncidents({ workspace: resolved.workspace, actor, }).catch((error) => { logger.warn("slack home incidents failed", { error, teamId }); return undefined; }), }); } else { const url = await linkAccountUrl(config, { workspaceId: resolved.workspace.id, teamId, slackUserId: userId, }); await publishLinkAccountView(slack, userId, url); } } catch (err) { logger.error("slack failed to publish home view", { error: err, teamId }); } return; }
// Only fires while the app is still on `assistant_view`. Delete this branch // once the `agent_view` manifest is live — the agent experience greets from // `app_home_opened` above instead. if (event.type === "assistant_thread_started") { const teamId = body.team_id; const thread = event.assistant_thread; if (!teamId || !thread?.user_id) return; const resolved = await resolveWorkspace(teamId); if (!resolved) return; const slack = new WebClient(resolved.botToken); const slackUserId = thread.user_id; try { const actor = await requireSlackMember({ workspace: resolved.workspace, teamId, slackUserId, slack, }); if (!actor) { await sendLinkCard({ config, workspaceId: resolved.workspace.id, teamId, slackUserId, post: (message) => slack.chat.postMessage({ channel: thread.channel_id, thread_ts: thread.thread_ts, ...message, }), }); return; } await greetOnce({ slack, teamId, userId: slackUserId, channel: thread.channel_id, threadTs: thread.thread_ts, }); } catch (err) { logger.error("slack failed to greet in new thread", { error: err, teamId, }); } return; }
// What the user is looking at, remembered for the next turn. Nothing is // read here — this only records where they are. if (event.type === "app_context_changed") { const teamId = body.team_id; const userId = contextUserId(body.authorizations); if (!teamId || !userId) return;
const channelId = contextChannelId(event.context?.entities); if (channelId) { await rememberContext(teamId, userId, channelId); } else { await forgetContext(teamId, userId); } return; }
// A person renamed the thread, so the name is theirs now — record it so no // later turn overwrites it. if (event.type === "agent_session_title_changed") { if (!body.team_id || !event.channel || !event.thread_ts) return; await markThreadTitled(body.team_id, event.channel, event.thread_ts); logger.info("slack thread renamed by user", { teamId: body.team_id, channel: event.channel, threadTs: event.thread_ts, }); return; }
// The user pressed stop. Slack has already halted any streamed message and // will not clear the session status itself, so both are on us. if (event.type === "agent_session_stopped") { const teamId = body.team_id; const channel = event.channel; const threadTs = event.thread_ts; if (!teamId || !channel || !threadTs) return;
const wasRunning = abortTurn(channel, threadTs); await broadcastStop(channel, threadTs).catch((error) => logger.warn("slack failed to broadcast stop", { error, teamId }), ); logger.info("slack turn stopped by user", { teamId, channel, threadTs, wasRunning, });
// Cleared here as well as in the aborted turn's own cleanup: the turn may // be running on another instance, or already be gone. const resolved = await resolveWorkspace(teamId); if (!resolved) return; await setSessionStatus( new WebClient(resolved.botToken), channel, threadTs, "active", ).catch((err: unknown) => logger.warn("slack failed to clear status after stop", { error: err, channel, teamId, }), ); return; }
if (event.type !== "app_mention" && event.type !== "message") return; if (event.type === "message" && event.bot_id) return;
// The agent pane is the app's DM: every message there is addressed to us, // so no mention is required. const isAgentThread = event.channel_type === "im";
const ignoredSubtypes = [ "channel_join", "channel_leave", "channel_topic", "channel_purpose", "channel_name", ]; if (event.subtype && ignoredSubtypes.includes(event.subtype)) return; // In the agent pane, subtypes are the thread root (`assistant_app_thread`), // edits and deletions — only plain user messages start a turn. if (isAgentThread && event.subtype) return;
const teamId = body.team_id; if (!teamId || !event.channel || !event.ts) return;
// A single mention arrives as BOTH an `app_mention` and a `message.*` event // (distinct event_ids, same message ts), so the event_id dedup above doesn't // catch the pair. Dedup on the message identity so we only respond once. if (await isDuplicate(`msg:${event.channel}:${event.ts}`)) return;
const resolved = await resolveWorkspace(teamId); if (!resolved) { logger.warn("slack integration not found", { teamId }); return; }
const slack = new WebClient(resolved.botToken); const botUserId = resolved.botUserId; const threadTs = event.thread_ts ?? event.ts;
// Fetched early only when needed to decide whether to answer; reused below // so the agent sees the same thread. let prefetchedThread: ThreadMessage[] | undefined; if ( !isAgentThread && event.type === "message" && !event.text?.includes(`<@${botUserId}>`) ) { if (!event.thread_ts) return; try { prefetchedThread = await fetchThread( slack, event.channel, event.thread_ts, ); } catch (err) { logger.warn("slack failed to fetch thread for untagged reply", { error: err, channel: event.channel, teamId, }); return; } if ( !isAnswerToAgent( prefetchedThread, { ts: event.ts, user: event.user }, botUserId, ) ) { return; } }
logger.info("slack event received", { teamId, channel: event.channel, eventType: event.type, threadTs, user: event.user, agentThread: isAgentThread, });
const channel = event.channel; const slackUserId = event.user; if (!slackUserId) return; if (!slackAgentAllowed(resolved.workspace)) { if (await claimPlanNoticeWindow(teamId, slackUserId)) { const message = planRequiredMessage(config); await ( isAgentThread ? slack.chat.postMessage({ channel, thread_ts: threadTs, ...message }) : slack.chat.postEphemeral({ channel, user: slackUserId, thread_ts: event.thread_ts, ...message, }) ).catch((error) => logger.error("slack failed to send plan notice", { error, teamId }), ); } return; } const actor = await requireSlackMember({ workspace: resolved.workspace, teamId, slackUserId, slack, }); if (!actor) { await sendLinkCard({ config, workspaceId: resolved.workspace.id, teamId, slackUserId, post: (message) => isAgentThread ? slack.chat.postMessage({ channel, thread_ts: threadTs, ...message }) : slack.chat.postEphemeral({ channel, user: slackUserId, thread_ts: event.thread_ts, ...message, }), }).catch((error) => logger.error("slack failed to send link card", { error, teamId }), ); return; }
// Registered before the session is marked `processing`: a stop landing in // that window has to find a controller, or the turn runs on unstoppable. const turn = startTurn(event.channel, threadTs); let reply: Reply | undefined;
try { reply = await createReply({ slack, channel: event.channel, threadTs, teamId, userId: event.user, isAgentThread, });
if (turn.signal.aborted) { logger.info("slack turn stopped before it started", { teamId, channel: event.channel, threadTs, }); await reply.stopped(); return; }
// the thread root: events the turn writes link back to where it was asked const permalink = permalinkFor(slack, event.channel, threadTs); let thread: ThreadMessage[] = []; if (prefetchedThread) { thread = prefetchedThread; } else if (event.thread_ts) { thread = await fetchThread(slack, event.channel, event.thread_ts); } else { thread = [{ user: event.user, text: event.text, ts: event.ts }]; }
// Only in the pane: in a channel the agent is already reading the thread // it was mentioned in, and "what you're looking at" is that same channel. const contextChannel = isAgentThread && event.user ? await recallContext(teamId, event.user) : undefined; const context = contextChannel ? channelContextTooling({ slack, channelId: contextChannel }) : undefined; const history = isAgentThread ? undefined : await fetchChannelHistory(slack, event.channel, { limit: event.thread_ts ? HISTORY_BEFORE_THREAD : HISTORY_BEFORE_MENTION, before: threadTs, }).catch((error) => { logger.info("slack channel history unavailable", { error, channel: event.channel, }); return []; }); const channelTools = isAgentThread ? context?.tools : channelHistoryTooling({ slack, channelId: event.channel }); const linksNote = await linkedMessagesNote({ slack, text: event.text, requesterId: event.user, }); const memoryNote = createdNote(await listCreated(threadTs).catch(() => [])); const linkTools = event.user ? slackLinkTools({ slack, requesterId: event.user }) : undefined; const incidentNote = await boundIncidentNote({ workspace: resolved.workspace, actor, teamId, channelId: isAgentThread ? contextChannel : event.channel, });
logger.info("slack agent invoked", { teamId, channel: event.channel, threadTs, messageCount: thread.length, contextChannel, });
const result = await runAgent( resolved.workspace, thread, botUserId, event.text, { ...actor, permalink: await permalink }, { events: reply.progress, signal: turn.signal, tools: { ...channelTools, ...linkTools }, contextNote: [ context?.contextNote, history && channelHistoryNote(event.channel, history), linksNote, memoryNote, incidentNote, ] .filter(Boolean) .join("\n\n") || undefined, }, );
if (result.aborted) { logger.info("slack turn abandoned after stop", { teamId, channel: event.channel, threadTs, }); await reply.stopped(); return; }
logger.info("slack agent completed", { teamId, channel: event.channel, threadTs, toolCalls: result.toolResults.map((tr) => tr.toolName), finishReason: result.finishReason, stepCount: result.stepCount, hitStepLimit: result.hitStepLimit, });
// One card per draft. A turn can draft several changes (rename a report // *and* post an update to it); dropping all but the first would leave the // user with nothing to click for the rest. When the model drafts the same // thing twice, its last draft is the one it meant. const drafts = new Map<string, SlackToolDraft>(); for (const tr of result.toolResults) { if (!isSlackToolDraft(tr.result)) continue; const key = draftKey({ toolName: tr.result.toolName, input: tr.result.input, }); drafts.delete(key); drafts.set(key, tr.result); } const firstDraft = drafts.values().next().value;
if (firstDraft) { for (const draft of drafts.values()) { logger.info("slack confirmation requested", { teamId, channel: event.channel, threadTs, toolName: draft.toolName, }); await handleConfirmation( slack, reply, event.channel, threadTs, event.user ?? "", resolved.workspace.id, teamId, draft, ); } } else { // No draft means no card. Distinguish a legitimate text answer from the // model drafting a change in prose and asking for permission instead of // calling the tool — the latter strands the user with nothing to click. if (looksLikeUncardedDraft(result.text)) { logger.warn("slack draft proposed without a card", { teamId, channel: event.channel, threadTs, finishReason: result.finishReason, stepCount: result.stepCount, hitStepLimit: result.hitStepLimit, readToolCalls: result.toolResults.map((tr) => tr.toolName), }); } if (result.text) { await reply.answer(result.text); } else { await reply.send({ text: "Done!" }); } logger.info("slack response sent", { teamId, channel: event.channel, threadTs, }); }
await titleThread({ slack, channel: event.channel, threadTs, teamId, workspaceId: resolved.workspace.id, isAgentThread, draft: firstDraft, userText: event.text, }); } catch (err) { logger.error("slack agent error", { error: err, channel: event.channel, teamId, threadTs, }); await reply ?.send({ text: ":x: Something went wrong. Please try again." }) .catch((sendErr: unknown) => { logger.error("slack failed to send error message", { error: sendErr, channel: event.channel, threadTs, }); }); } finally { endTurn(event.channel, threadTs, turn); await reply?.finish?.(); }}
function statusReportIdOf(input: unknown): number | undefined { if (typeof input !== "object" || input === null) return undefined; const id = (input as { statusReportId?: unknown }).statusReportId; return typeof id === "number" ? id : undefined;}
/** * Names the thread after its subject, so the agent pane's timeline reads as a * list of incidents rather than a list of operations. * * Confined to the agent pane: that timeline is the whole payoff, and Slack * documents `agents.sessions.rename` as also renaming the channel for session * channels — not something to risk on a shared incident channel. * * Cosmetic, and last: a failure here must never cost the user their answer. */async function titleThread(args: { slack: WebClient; channel: string; threadTs: string; teamId: string; workspaceId: number; isAgentThread: boolean; draft?: { toolName: string; input: unknown }; userText?: string;}): Promise<void> { const { slack, channel, threadTs, teamId, workspaceId, isAgentThread, draft, userText, } = args; if (!isAgentThread) return;
try { if (await isThreadTitled(teamId, channel, threadTs)) return;
// Only looked up once, and only for a draft that acts on an existing // report — `add_update` and `resolve` carry an id but no title. let reportTitle: string | undefined; const reportId = draft && statusReportIdOf(draft.input); if (reportId !== undefined) { const link = await getStatusReportLink(workspaceId, reportId); reportTitle = link?.title; }
const title = buildThreadTitle({ draft, reportTitle, userText }); if (!title) return;
await renameThread({ slack, channel, threadTs, title, teamId }); } catch (err) { logger.warn("slack failed to title the thread", { error: err, channel, teamId, }); }}
/** * Where the agent's output goes, in descending order of how much the user gets * to see while they wait: a streamed message that fills in as the model writes, * with a task entry per tool call; or a native "working" status on the thread. */interface Reply { /** * Delivers the agent's free-text answer and returns the ts of the message * holding it. When the answer was streamed it is already on screen, and this * only finalizes the message. */ answer(text: string): Promise<string>; /** Writes a message of our own — a confirmation card, or an error. */ send(message: { text: string; blocks?: Block[] }): Promise<string>; /** * Confirms the turn ended because the user stopped it, leaving whatever was * already written in place. */ stopped(): Promise<void>; /** Progress to report while the turn runs, when the surface can show it. */ progress?: AgentEvents; /** Runs once the turn is over, whether it succeeded or not. */ finish?: () => Promise<void>;}
const STOPPED_NOTICE = "_Stopped._";
/** * Registry tools are named `verb_noun` (`list_status_pages`, * `get_monitor_status`). A task list reads better as an activity, so the verb * becomes a gerund and the rest is left as words. */const TOOL_ACTIVITY: Record<string, string> = { list: "Reading", get: "Reading", search: "Searching", create: "Drafting", add: "Drafting", update: "Drafting", resolve: "Drafting",};
export function toolTaskTitle(toolName: string): string { const [verb, ...rest] = toolName.split("_"); const subject = rest.join(" "); const activity = TOOL_ACTIVITY[verb]; if (!activity || !subject) return toolName.replace(/_/g, " "); return `${activity} ${subject}`;}
function taskChunk( id: string, toolName: string, status: "in_progress" | "complete",) { return { type: "task_update" as const, id, title: toolTaskTitle(toolName), status, };}
/** * Opens a stream for the turn, or returns undefined when this surface can't * carry one. Whether the *workspace* allows streaming only shows up on the * first append, mid-turn — `streamingReply` handles that failure. */function createStreamer(args: { slack: WebClient; channel: string; threadTs: string; teamId: string; userId: string | undefined; isAgentThread: boolean;}) { const { slack, channel, threadTs, teamId, userId, isAgentThread } = args; // Older @slack/web-api has no streaming support. if (typeof slack.chatStream !== "function") return undefined; // Outside a DM, Slack needs to know who the streamed message is for. if (!isAgentThread && !userId) return undefined; try { return slack.chatStream({ channel, thread_ts: threadTs, task_display_mode: "timeline", recipient_user_id: userId, recipient_team_id: teamId, }); } catch (err) { logger.warn("slack could not open a stream", { error: err, channel, teamId, }); return undefined; }}
/** * Streams the answer as the model writes it. Every stream call is best-effort: * a workspace without streaming enabled only fails on the first append, by * which point the turn is already running, so a failure latches into `broken` * and the answer is posted (or the half-written message rewritten) instead. */function streamingReply(args: { slack: WebClient; streamer: NonNullable<ReturnType<typeof createStreamer>>; channel: string; threadTs: string; teamId: string; finishSession?: () => Promise<void>;}): Reply { const { slack, streamer, channel, threadTs, teamId, finishSession } = args; const post = postInThread(slack, channel, threadTs);
let appended = false; let streamedText = false; let broken = false; let streamClosed = false;
const attempt = async (fn: () => Promise<void>) => { if (broken) return; try { await fn(); appended = true; } catch (err) { broken = true; logger.warn("slack stream failed, falling back to a posted message", { error: err, channel, teamId, }); } };
/** * Finalizes the streamed message, if one was ever opened — including after a * later append failed, or Slack leaves that message streaming for good. */ const closeStream = async (): Promise<string | undefined> => { if (streamClosed || !appended) return streamer.ts; streamClosed = true; try { await streamer.stop(); } catch (err) { broken = true; logger.warn("slack failed to stop the stream", { error: err, channel, teamId, }); } return streamer.ts; };
return { progress: { onTextDelta: (delta) => attempt(async () => { await streamer.append({ markdown_text: delta }); streamedText = true; }), onToolCall: ({ id, toolName }) => attempt(async () => { await streamer.append({ chunks: [taskChunk(id, toolName, "in_progress")], }); }), onToolResult: ({ id, toolName }) => attempt(async () => { await streamer.append({ chunks: [taskChunk(id, toolName, "complete")], }); }), }, async answer(text) { const ts = await closeStream(); if (streamedText && !broken) return ts ?? ""; // The stream never carried the answer — nothing was streamed, or it // broke partway. Put the whole answer on screen, rewriting the // half-written message when there is one. const message = buildAnswerMessage(text); if (ts) { await slack.chat.update({ channel, ts, ...message }); return ts; } return post(message); }, async send(message) { // The card is a message of its own: updating the streamed one in place // would wipe the answer Slack has already rendered. await closeStream(); return post(message); }, async stopped() { if (streamClosed || !appended) { await post({ text: STOPPED_NOTICE }); return; } streamClosed = true; try { // Closes the partial answer with the notice attached. Slack halts the // stream on its side when the user presses stop, so this often fails — // the partial message is already final, which is the point. await streamer.stop({ markdown_text: `\n\n${STOPPED_NOTICE}` }); } catch (err) { logger.info("slack stream already closed by the stop request", { error: err, channel, teamId, }); } }, async finish() { await closeStream(); await finishSession?.(); }, };}
/** * Picks the richest delivery this surface supports. The session status is * independent of streaming — a streamed turn still marks the thread as * working, so Slack shows the loading state and the stop button. */async function createReply(args: { slack: WebClient; channel: string; threadTs: string; teamId: string; userId: string | undefined; isAgentThread: boolean;}): Promise<Reply> { const { slack, channel, threadTs, teamId, userId } = args;
const session = await acknowledgeWithSession( slack, channel, threadTs, teamId, userId, );
// The status is the loading indicator, not the delivery: a streamed turn // still needs it when agent sessions aren't available. if (!session) { // Best-effort: a missing status only loses the loading indicator. await setAssistantStatus(slack, channel, threadTs, "is thinking...").catch( (err: unknown) => logger.warn("slack failed to set assistant status", { error: err, channel, teamId, }), ); }
const streamer = createStreamer(args); if (streamer) { return streamingReply({ slack, streamer, channel, threadTs, teamId, finishSession: session?.finish, }); }
return session ?? acknowledgeWithStatus(slack, channel, threadTs);}
function postInThread( slack: WebClient, channel: string, threadTs: string,): Reply["send"] { return async ({ text, blocks }) => { try { const res = await slack.chat.postMessage({ channel, thread_ts: threadTs, text, blocks, }); if (!res.ts) throw new Error("chat.postMessage returned no ts"); return res.ts; } catch (err) { if (!isSlackPlatformError(err, "cannot_reply_to_message")) throw err; // The parent message can't host a thread — answer at the top level // rather than losing the reply entirely. logger.warn("slack cannot reply to message, falling back to top-level", { channel, threadTs, }); const res = await slack.chat.postMessage({ channel, text, blocks }); if (!res.ts) throw new Error("chat.postMessage returned no ts"); return res.ts; } };}
/** * Preferred acknowledgement: mark the thread's agent session as `processing` * so Slack shows the agent working — no placeholder message — and hand it * back as `active` once we've answered. Returns undefined when the workspace * doesn't support agent sessions, so the caller falls back to the older * indicators. */async function acknowledgeWithSession( slack: WebClient, channel: string, threadTs: string, teamId: string, userId: string | undefined,): Promise<Reply | undefined> { try { await setSessionStatus(slack, channel, threadTs, "processing", userId); } catch (err) { logger.info("slack agent session unavailable, falling back", { error: err, channel, teamId, }); return; } const send = postInThread(slack, channel, threadTs); return { send, answer: (text) => send(buildAnswerMessage(text)), stopped: async () => { await send({ text: STOPPED_NOTICE }); }, async finish() { await setSessionStatus(slack, channel, threadTs, "active").catch( (err: unknown) => logger.warn("slack failed to reset agent session status", { error: err, channel, teamId, }), ); }, };}
// The status was already set by `createReply`; Slack clears it as soon as the// app posts in the thread, so there is no matching "clear" call.function acknowledgeWithStatus( slack: WebClient, channel: string, threadTs: string,): Reply { const send = postInThread(slack, channel, threadTs); return { send, answer: (text) => send(buildAnswerMessage(text)), stopped: async () => { await send({ text: STOPPED_NOTICE }); }, };}
async function handleConfirmation( slack: WebClient, reply: Reply, channel: string, threadTs: string, userId: string, workspaceId: number, teamId: string, draft: SlackToolDraft,) { const tool = getRegistryTool(draft.toolName); if (!tool) { logger.error("slack: registry tool not found", { toolName: draft.toolName, }); await reply.send({ text: ":x: Something went wrong. Please try again." }); return; }
const payload: PendingPayload = { toolName: draft.toolName, input: draft.input, }; const text = getConfirmationText({ tool, input: draft.displayInput });
// findByThread + replace isn't atomic on its own — two concurrent // events on the same thread could both see `existing` and race on // replace. The Redis event dedup at the top of this file suppresses // duplicate deliveries across instances, and Slack throttles per thread. const existing = await findByThread(threadTs, payload); if (existing) { await replace(existing.id, payload);
const blocks = await buildConfirmationBlocks({ actionId: existing.id, tool, input: draft.displayInput, resolvers: makeRefResolvers(workspaceId), }); await reply.send({ text, blocks }); await slack.chat.update({ channel, ts: existing.messageTs, text, blocks, }); } else { // The card's buttons carry the action id, and the stored action carries // the card's ts — so write the text first to learn the ts, then attach // the buttons. const messageTs = await reply.send({ text }); const actionId = await store({ workspaceId, teamId, channelId: channel, threadTs, messageTs, userId, payload, });
const blocks = await buildConfirmationBlocks({ actionId, tool, input: draft.displayInput, resolvers: makeRefResolvers(workspaceId), }); await slack.chat.update({ channel, ts: messageTs, text, blocks }); }}