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
at incident-api
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387import { getLogger } from "@logtape/logtape";import { isFeatureEnabled } from "@openstatus/services";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, contextChannelId, contextUserId, forgetContext, recallContext, rememberContext,} from "./channel-context";import type { SlackConfig, SlackEnv } from "./config";import { makeRefResolvers } from "./confirmation-card";import { draftKey, findByThread, 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 { buildThreadTitle, isThreadTitled, markThreadTitled, renameThread,} from "./thread-title";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(), 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;}
// 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 });}
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; }
// 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 = isFeatureEnabled(resolved.workspace, "incident-management") && 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; }
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 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, { events: reply.progress, signal: turn.signal, tools: context?.tools, contextNote: [context?.contextNote, 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 }); }}