diff --git a/apps/server/slack-manifest.json b/apps/server/slack-manifest.json index 49dd6c72..22cd0acb 100644 --- a/apps/server/slack-manifest.json +++ b/apps/server/slack-manifest.json @@ -6,8 +6,8 @@ "long_description": "openstatus is an agent for incident communication. Open it from the Slack top bar, or mention @openstatus in any channel, and it turns the conversation into status page updates — no context switching required.\n\n How it works:\n :one: Open openstatus in the agent pane and describe the issue, or mention @openstatus in an incident channel or thread\n :two: The agent reads the conversation, queries your status pages, and drafts a status update\n :three: Review the proposed action and click Approve, Approve & Notify, or Cancel\n\n What you can do:\n :white_check_mark: Create status reports from natural conversation\n :white_check_mark: Post progress updates as incidents evolve\n :white_check_mark: Resolve incidents when the issue is fixed\n :white_check_mark: Schedule maintenance windows in plain language\n :white_check_mark: Notify your status page subscribers with one click\n\n The agent understands context — tag it in an ongoing thread and it will synthesize the discussion into a clear, public-facing status update. It infers incident status automatically: \"we found the root cause\" becomes Identified, \"it's fixed\" becomes Resolved.\n\n AI disclaimer: openstatus uses large language models (LLMs) to summarize conversations, draft status updates, and infer incident status. AI-generated content may sometimes be inaccurate, incomplete, or misleading. Always review every draft before approving it — nothing is posted to your status page until you explicitly confirm. Only the person who triggered the action can approve it, and you can edit or cancel any draft." }, "features": { - "assistant_view": { - "assistant_description": "Drafts, updates and resolves status reports and maintenance windows. Nothing is published until you approve it.", + "agent_view": { + "agent_description": "Drafts, updates and resolves status reports and maintenance windows. Nothing is published until you approve it.", "suggested_prompts": [ { "title": "Create a status report", @@ -70,9 +70,11 @@ "event_subscriptions": { "request_url": "https://api.openstatus.dev/slack/events", "bot_events": [ + "agent_session_stopped", + "agent_session_title_changed", + "app_context_changed", "app_home_opened", "app_mention", - "assistant_thread_started", "message.channels", "message.groups", "message.im" diff --git a/apps/server/src/libs/test/doubles/slack-agent.mock.ts b/apps/server/src/libs/test/doubles/slack-agent.mock.ts index c6d5c2fb..9bcaef3c 100644 --- a/apps/server/src/libs/test/doubles/slack-agent.mock.ts +++ b/apps/server/src/libs/test/doubles/slack-agent.mock.ts @@ -2,13 +2,25 @@ // --import-map so handler tests don't run the real agent. import { slackTestState } from "./slack-test-state.ts"; -export const runAgent = () => { - if (slackTestState.runAgentOverride) return slackTestState.runAgentOverride(); +// Signature mirrors the real runAgent so overrides can drive `events` and +// observe the abort signal. +export const runAgent = ( + _workspace?: unknown, + _thread?: unknown, + _botUserId?: unknown, + _userText?: unknown, + _origin?: unknown, + options?: unknown, +) => { + if (slackTestState.runAgentOverride) { + return slackTestState.runAgentOverride(options); + } return Promise.resolve({ text: "Here is my response", toolResults: [], finishReason: "stop", stepCount: 1, hitStepLimit: false, + aborted: false, }); }; 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 faec30f2..efdaf0d7 100644 --- a/apps/server/src/libs/test/doubles/slack-test-state.ts +++ b/apps/server/src/libs/test/doubles/slack-test-state.ts @@ -11,8 +11,15 @@ export interface SlackTestState { updateOverride: Override; postEphemeralOverride: Override; sessionStatusOverride: Override; - runAgentOverride: (() => Promise) | null; + renameOverride: Override; + /** Set to false to simulate a workspace/SDK without message streaming. */ + chatStreamEnabled: boolean; + /** Make appends start throwing once this many have succeeded. */ + streamAppendFailAfter: number | null; + /** Receives runAgent's options, so a test can drive the stream or abort. */ + runAgentOverride: ((options?: unknown) => Promise) | null; repliesImpl: () => Promise; + historyImpl: () => Promise; } const g = globalThis as Record; @@ -24,11 +31,18 @@ if (!g.__slackTestState) { updateOverride: null, postEphemeralOverride: null, sessionStatusOverride: null, + renameOverride: null, + chatStreamEnabled: true, + streamAppendFailAfter: null, runAgentOverride: null, repliesImpl: () => Promise.resolve({ messages: [{ user: "U1", text: "test message", ts: "1.1" }], }), + historyImpl: () => + Promise.resolve({ + messages: [{ user: "U1", text: "channel message", ts: "1.1" }], + }), } satisfies SlackTestState; } 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 a9e5b221..7f6ae3fb 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,8 +20,43 @@ export class WebClient { return Promise.resolve(); }, }; + // Mirrors ChatStreamer: `ts` is undefined until the first append or stop. + chatStream = (args: Record) => { + if (!s.chatStreamEnabled) { + throw new Error("chat streaming is not enabled for this workspace"); + } + s.calls.push({ method: "chatStream", args }); + let ts: string | undefined; + let appends = 0; + return { + get ts() { + return ts; + }, + append: (a: Record) => { + if ( + s.streamAppendFailAfter !== null && + appends >= s.streamAppendFailAfter + ) { + return Promise.reject(new Error("stream append failed")); + } + appends++; + ts = "stream.ts"; + s.calls.push({ method: "stream.append", args: a }); + return Promise.resolve(null); + }, + stop: (a?: Record) => { + ts = "stream.ts"; + s.calls.push({ method: "stream.stop", args: a ?? {} }); + return Promise.resolve({ ok: true, ts }); + }, + }; + }; conversations = { replies: () => s.repliesImpl(), + history: (args: Record) => { + s.calls.push({ method: "conversations.history", args }); + return s.historyImpl(); + }, }; agents = { sessions: { @@ -30,6 +65,11 @@ export class WebClient { s.calls.push({ method: "agents.sessions.setStatus", args }); return Promise.resolve({ ok: true }); }, + rename: (args: Record) => { + if (s.renameOverride) return s.renameOverride(args); + s.calls.push({ method: "agents.sessions.rename", args }); + return Promise.resolve({ ok: true }); + }, }, }; assistant = { @@ -38,10 +78,6 @@ export class WebClient { s.calls.push({ method: "assistant.threads.setStatus", args }); return Promise.resolve({ ok: true }); }, - setSuggestedPrompts: (args: Record) => { - s.calls.push({ method: "assistant.threads.setSuggestedPrompts", args }); - return Promise.resolve({ ok: true }); - }, }, }; views = { diff --git a/apps/server/src/routes/slack/agent.ts b/apps/server/src/routes/slack/agent.ts index da99037e..2933fb99 100644 --- a/apps/server/src/routes/slack/agent.ts +++ b/apps/server/src/routes/slack/agent.ts @@ -1,7 +1,7 @@ import type { Workspace } from "@openstatus/db/src/schema/workspaces/validation"; import type { ServiceContext } from "@openstatus/services"; -import { generateText, stepCountIs } from "ai"; -import type { ModelMessage } from "ai"; +import { stepCountIs, streamText } from "ai"; +import type { ModelMessage, Tool } from "ai"; import { tb } from "@/libs/clients"; @@ -32,6 +32,30 @@ interface AgentResult { stepCount: number; /** True when the model used every allowed step — a draft may have been cut off. */ hitStepLimit: boolean; + /** True when the user stopped the turn; `text` is then whatever was written. */ + aborted: boolean; +} + +/** + * Progress reported as the turn runs, so the surface can show the answer + * arriving and the tools being used rather than a spinner. Every callback is + * awaited inside the stream loop: a slow one slows the turn, and a throwing + * one aborts it, so implementations swallow their own failures. + */ +export interface AgentEvents { + onTextDelta(delta: string): Promise; + onToolCall(call: { id: string; toolName: string }): Promise; + onToolResult(result: { id: string; toolName: string }): Promise; +} + +export interface AgentOptions { + events?: AgentEvents; + /** Aborted when the user presses Slack's stop button. */ + signal?: AbortSignal; + /** Surface-only tools, merged over the registry's. */ + tools?: Record; + /** Appended to the system prompt: what the user is looking at right now. */ + contextNote?: string; } function convertThreadToMessages( @@ -61,6 +85,7 @@ export async function runAgent( botUserId: string, userText?: string, origin?: { slackUserId: string; teamId: string | undefined }, + options?: AgentOptions, ): Promise { const ctx: ServiceContext = { workspace, @@ -71,7 +96,7 @@ export async function runAgent( }, tb, }; - const tools = buildSlackTools(ctx); + const tools = buildSlackTools(ctx, options?.tools); let messages = convertThreadToMessages(thread, botUserId); if (messages.length === 0 && userText) { @@ -85,29 +110,93 @@ export async function runAgent( finishReason: "unknown", stepCount: 0, hitStepLimit: false, + aborted: false, }; } - const result = await generateText({ + const { events, signal, contextNote } = options ?? {}; + + const result = streamText({ model: MODEL, - system: buildSystemPrompt(workspace.name ?? "Unknown"), + system: buildSystemPrompt(workspace.name ?? "Unknown", contextNote), messages, tools, stopWhen: stepCountIs(MAX_STEPS), + abortSignal: signal, }); + // The text is accumulated here rather than read from `result.text` at the + // end, because a stopped turn has no end to wait for — this is what the + // user was shown before they stopped it. + let text = ""; + let aborted = false; + + // `fullStream` has to be drained for the turn to run to completion, whether + // or not anyone is listening. A model error surfaces when the settled + // promises below are awaited, so it still reaches the caller's catch. + try { + for await (const part of result.fullStream) { + if (signal?.aborted) { + aborted = true; + break; + } + switch (part.type) { + case "text-delta": + text += part.text; + await events?.onTextDelta(part.text); + break; + case "tool-call": + await events?.onToolCall({ + id: part.toolCallId, + toolName: part.toolName, + }); + break; + case "tool-result": + await events?.onToolResult({ + id: part.toolCallId, + toolName: part.toolName, + }); + break; + case "abort": + aborted = true; + break; + default: + break; + } + if (aborted) break; + } + } catch (err) { + // An abort surfaces as a rejection in some providers and as an `abort` + // part in others. Anything else is a real failure. + if (!signal?.aborted) throw err; + aborted = true; + } + + if (aborted) { + return { + text, + toolResults: [], + finishReason: "abort", + stepCount: 0, + hitStepLimit: false, + aborted: true, + }; + } + + const steps = await result.steps; const toolResults: AgentResult["toolResults"] = []; - for (const step of result.steps) { + for (const step of steps) { for (const tc of step.toolResults) { toolResults.push({ toolName: tc.toolName, result: tc.output }); } } return { - text: result.text, + text: await result.text, toolResults, - finishReason: result.finishReason, - stepCount: result.steps.length, - hitStepLimit: result.steps.length >= MAX_STEPS, + finishReason: await result.finishReason, + stepCount: steps.length, + hitStepLimit: steps.length >= MAX_STEPS, + aborted: false, }; } diff --git a/apps/server/src/routes/slack/assistant.ts b/apps/server/src/routes/slack/assistant.ts index c58db42b..ea87f5ac 100644 --- a/apps/server/src/routes/slack/assistant.ts +++ b/apps/server/src/routes/slack/assistant.ts @@ -1,42 +1,96 @@ -import type { WebClient } from "@slack/web-api"; +import { getLogger } from "@logtape/logtape"; +import type { KnownBlock, WebClient } from "@slack/web-api"; -// Shown when a user opens a new thread in the agent pane. Slack caps a thread -// at four prompts; titles render as chips, `message` is what gets sent. -export const SUGGESTED_PROMPTS = [ - { - title: "Create a status report", - message: - "We're seeing elevated errors on the API — create a status report.", - }, - { - title: "Open status reports", - message: "Which status reports are currently open?", - }, +import { redis } from "@/libs/clients"; + +import { DOCS_URL } from "./home"; + +const logger = getLogger("api-server"); + +// Suggested prompts live in `features.agent_view.suggested_prompts` in +// slack-manifest.json. The agent experience renders them at the top of the +// Messages tab rather than inside a thread, so there is no runtime call to +// make and no second copy here to drift from the manifest. + +/** + * First contact. Short, because it sits above the manifest's prompt chips, + * which already show what to ask for. It exists to say the one thing nobody + * should have to discover by trying it: the agent drafts, and nothing reaches + * the status page without a click. + */ +const WELCOME_TEXT = + "I turn incident conversations into status page updates. I draft, you approve — nothing is published until you click Approve."; + +const WELCOME_BLOCKS: KnownBlock[] = [ { - title: "Schedule maintenance", - message: "Schedule a maintenance window next Tuesday from 2-3 PM UTC.", + type: "section", + text: { + type: "mrkdwn", + text: "I turn incident conversations into status page updates.\n\nTell me what's happening, or mention *@openstatus* in a channel and I'll read the thread. *I draft, you approve* — nothing reaches your status page until you click *Approve*. I can also answer questions about your monitors and the openstatus docs.", + }, }, { - title: "Upcoming maintenance", - message: "What maintenance windows are coming up?", + type: "context", + elements: [ + { + type: "mrkdwn", + text: `Drafts are AI-generated — read them before approving. <${DOCS_URL}|Documentation>`, + }, + ], }, ]; -export async function startAssistantThread( - slack: WebClient, - channel: string, - threadTs: string, -): Promise { - await slack.assistant.threads.setSuggestedPrompts({ - channel_id: channel, - thread_ts: threadTs, - title: "What do you need to communicate?", - prompts: SUGGESTED_PROMPTS, +/** + * Whether this person has met the agent before. The agent experience has no + * per-thread empty state to greet into — opening the Messages tab is the only + * hook, and it fires every time — so the greeting is once per person. + */ +const GREETED_PREFIX = "slack:greeted:"; +const GREETED_TTL_SECONDS = 365 * 24 * 60 * 60; + +function greetedKey(teamId: string, userId: string): string { + return `${GREETED_PREFIX}${teamId}:${userId}`; +} + +export async function hasBeenGreeted( + teamId: string, + userId: string, +): Promise { + return (await redis.get(greetedKey(teamId, userId))) !== null; +} + +/** + * Greets someone the first time they open the agent, and never again. + * + * `threadTs` is set only on the legacy `assistant_thread_started` path, where + * the greeting belongs in the thread that was just opened. + */ +export async function greetOnce(args: { + slack: WebClient; + teamId: string; + userId: string; + channel: string; + threadTs?: string; +}): Promise { + const { slack, teamId, userId, channel, threadTs } = args; + if (await hasBeenGreeted(teamId, userId)) return; + + await slack.chat.postMessage({ + channel, + ...(threadTs ? { thread_ts: threadTs } : {}), + text: WELCOME_TEXT, + blocks: WELCOME_BLOCKS, + }); + + await redis.set(greetedKey(teamId, userId), "1", { + ex: GREETED_TTL_SECONDS, }); + logger.info("slack greeted user", { teamId, userId, channel }); } // Slack clears the status as soon as the app posts in the thread, so there is -// no matching "clear" call on the success path. +// no matching "clear" call on the success path. Superseded by +// `setSessionStatus` — kept for workspaces the agent session API rejects. export async function setAssistantStatus( slack: WebClient, channel: string, diff --git a/apps/server/src/routes/slack/background.ts b/apps/server/src/routes/slack/background.ts new file mode 100644 index 00000000..ed01bf31 --- /dev/null +++ b/apps/server/src/routes/slack/background.ts @@ -0,0 +1,47 @@ +import { getLogger } from "@logtape/logtape"; + +const logger = getLogger("api-server"); + +/** + * Work that runs after the HTTP response has been sent. + * + * Slack gives us 3 seconds to acknowledge an event, an interaction or a slash + * command. Anything slower and the user sees a timeout warning — on a click + * that in fact succeeded. Approving a status report writes to the DB and then + * fans out to every subscriber, so it routinely outlives that window: ack + * first, do the work here, and report the outcome by updating the message (or + * via `response_url` for slash commands). + * + * The API server is a long-lived process, so a detached promise survives to + * completion; this is not safe on a request-scoped serverless runtime. + */ +const inFlight = new Set>(); + +export function runInBackground( + label: string, + work: () => Promise, + context: Record = {}, +): void { + const task: Promise = work() + .catch((error: unknown) => { + logger.error(`slack background task failed: ${label}`, { + error, + ...context, + }); + }) + .finally(() => { + inFlight.delete(task); + }); + inFlight.add(task); +} + +/** + * Resolves once every task started so far has settled. Tests use this instead + * of sleeping, so they assert on the finished side effects rather than on a + * timer that happens to be long enough. + */ +export async function settleBackgroundTasks(): Promise { + while (inFlight.size > 0) { + await Promise.all([...inFlight]); + } +} diff --git a/apps/server/src/routes/slack/blocks.ts b/apps/server/src/routes/slack/blocks.ts index e39c8905..0d47f328 100644 --- a/apps/server/src/routes/slack/blocks.ts +++ b/apps/server/src/routes/slack/blocks.ts @@ -4,6 +4,8 @@ import type { SummaryLine, } from "@openstatus/services/agent-tools"; +import { toMrkdwn } from "./mrkdwn"; + interface TextObject { type: "plain_text" | "mrkdwn"; text: string; @@ -29,6 +31,16 @@ interface ContextBlock { elements: TextObject[]; } +/** + * Renders standard markdown — the dialect the model actually writes — rather + * than Slack's mrkdwn, so tables, ordered lists and fenced code survive + * instead of being flattened by `toMrkdwn`. + */ +interface MarkdownBlock { + type: "markdown"; + text: string; +} + interface ButtonElement { type: "button"; text: TextObject; @@ -37,7 +49,31 @@ interface ButtonElement { style?: "primary" | "danger"; } -export type Block = SectionBlock | ActionsBlock | DividerBlock | ContextBlock; +export type Block = + | SectionBlock + | ActionsBlock + | DividerBlock + | ContextBlock + | MarkdownBlock; + +/** Slack caps all `markdown` blocks in one payload at 12,000 characters. */ +const MARKDOWN_BLOCK_LIMIT = 12_000; + +/** + * The agent's free-text answer as a Slack message. `text` carries the + * mrkdwn-converted copy — it is what notifications and screen readers use, and + * the fallback when the answer is too long for a `markdown` block. + */ +export function buildAnswerMessage(text: string): { + text: string; + blocks?: Block[]; +} { + const fallback = toMrkdwn(text); + if (!text.trim() || text.length > MARKDOWN_BLOCK_LIMIT) { + return { text: fallback }; + } + return { text: fallback, blocks: [{ type: "markdown", text }] }; +} /** * Action-id encoding. We need to round-trip both the pending action's id diff --git a/apps/server/src/routes/slack/channel-context.test.ts b/apps/server/src/routes/slack/channel-context.test.ts new file mode 100644 index 00000000..8c3fce7e --- /dev/null +++ b/apps/server/src/routes/slack/channel-context.test.ts @@ -0,0 +1,124 @@ +import { beforeEach, describe, expect, test } from "@openstatus/test-utils"; +// The double stands in for @slack/web-api via the test import map. +import { WebClient } from "@slack/web-api"; + +import { slackTestState } from "@/libs/test/doubles/slack-test-state"; + +import { + channelContextTooling, + contextChannelId, + contextUserId, + forgetContext, + recallContext, + rememberContext, +} from "./channel-context"; + +const redisStore = (globalThis as Record) + .__testRedisStore as Map; + +describe("reading the context payload", () => { + test("takes the authorizing human, not the bot", () => { + expect( + contextUserId([ + { user_id: "B9", is_bot: true }, + { user_id: "U1", is_bot: false }, + ]), + ).toBe("U1"); + }); + + test("falls back to any authorization carrying a user", () => { + expect(contextUserId([{ user_id: "U2" }])).toBe("U2"); + expect(contextUserId([])).toBeUndefined(); + expect(contextUserId(undefined)).toBeUndefined(); + }); + + test("takes the first channel, since entities are ranked by relevance", () => { + expect( + contextChannelId([ + { type: "slack#/types/user_id", value: "U9" }, + { type: "slack#/types/channel_id", value: "C_FIRST" }, + { type: "slack#/types/channel_id", value: "C_SECOND" }, + ]), + ).toBe("C_FIRST"); + }); + + test("has nothing to report for an empty context", () => { + expect(contextChannelId([])).toBeUndefined(); + expect(contextChannelId(undefined)).toBeUndefined(); + expect( + contextChannelId([{ type: "slack#/types/channel_id" }]), + ).toBeUndefined(); + }); +}); + +describe("the remembered channel", () => { + beforeEach(() => redisStore.clear()); + + test("round-trips per user and can be cleared", async () => { + expect(await recallContext("T1", "U1")).toBeUndefined(); + + await rememberContext("T1", "U1", "C_INCIDENT"); + expect(await recallContext("T1", "U1")).toBe("C_INCIDENT"); + expect(await recallContext("T1", "U2")).toBeUndefined(); + + await forgetContext("T1", "U1"); + expect(await recallContext("T1", "U1")).toBeUndefined(); + }); +}); + +describe("the read_slack_channel tool", () => { + beforeEach(() => { + slackTestState.calls = []; + slackTestState.historyImpl = () => + Promise.resolve({ + messages: [ + { user: "U2", text: "second", ts: "2.0" }, + { user: "U1", text: "first", ts: "1.0" }, + ], + }); + }); + + function tooling() { + return channelContextTooling({ + slack: new WebClient("xoxb-test"), + channelId: "C_INCIDENT", + }); + } + + test("names the channel in a form Slack renders", () => { + const { contextNote } = tooling(); + expect(contextNote).toContain("<#C_INCIDENT>"); + expect(contextNote).toContain("/invite @openstatus"); + }); + + test("returns the discussion oldest first", async () => { + const { tools } = tooling(); + // biome-ignore lint/suspicious/noExplicitAny: AI SDK tool execute shape + const result = (await (tools.read_slack_channel as any).execute({})) as { + messages: Array<{ text: string }>; + }; + + // Slack hands back newest first; a summary reads forwards. + expect(result.messages.map((m) => m.text)).toEqual(["first", "second"]); + }); + + test("hands a missing membership back as data, not an exception", async () => { + slackTestState.historyImpl = () => { + const err = new Error("An API error occurred: not_in_channel"); + Object.assign(err, { + code: "slack_webapi_platform_error", + data: { ok: false, error: "not_in_channel" }, + }); + return Promise.reject(err); + }; + + const { tools } = tooling(); + // biome-ignore lint/suspicious/noExplicitAny: AI SDK tool execute shape + const result = (await (tools.read_slack_channel as any).execute({})) as { + error: string; + }; + + // The model needs to explain this, so it must not blow up the turn. + expect(result.error).toBe("not_in_channel"); + }); +}); diff --git a/apps/server/src/routes/slack/channel-context.ts b/apps/server/src/routes/slack/channel-context.ts new file mode 100644 index 00000000..a9bb975b --- /dev/null +++ b/apps/server/src/routes/slack/channel-context.ts @@ -0,0 +1,142 @@ +import { getLogger } from "@logtape/logtape"; +import type { WebClient } from "@slack/web-api"; +import { type Tool, tool } from "ai"; +import { z } from "zod"; + +import { redis } from "@/libs/clients"; + +const logger = getLogger("api-server"); + +/** + * The channel the user is looking at, as reported by `app_context_changed`. + * + * Short-lived on purpose: it is a fact about right now, and acting on a + * channel someone glanced at half an hour ago would be worse than having no + * context at all. + */ +const CONTEXT_PREFIX = "slack:context:"; +const CONTEXT_TTL_SECONDS = 15 * 60; + +function contextKey(teamId: string, userId: string): string { + return `${CONTEXT_PREFIX}${teamId}:${userId}`; +} + +export async function rememberContext( + teamId: string, + userId: string, + channelId: string, +): Promise { + await redis.set(contextKey(teamId, userId), channelId, { + ex: CONTEXT_TTL_SECONDS, + }); +} + +export async function forgetContext( + teamId: string, + userId: string, +): Promise { + await redis.del(contextKey(teamId, userId)); +} + +export async function recallContext( + teamId: string, + userId: string, +): Promise { + return (await redis.get(contextKey(teamId, userId))) ?? undefined; +} + +/** The event carries no `user`; the authorizing human is in `authorizations`. */ +export function contextUserId( + authorizations: Array<{ user_id?: string; is_bot?: boolean }> | undefined, +): string | undefined { + if (!authorizations?.length) return undefined; + const human = authorizations.find((a) => a.is_bot === false && a.user_id); + return (human ?? authorizations.find((a) => a.user_id))?.user_id; +} + +/** + * The channel out of a context payload. Entities come ordered by relevance, + * so the first channel is the one being looked at. + */ +export function contextChannelId( + entities: Array<{ type?: string; value?: string }> | undefined, +): string | undefined { + return entities?.find((e) => e.type?.endsWith("channel_id") && e.value) + ?.value; +} + +const MAX_CONTEXT_MESSAGES = 50; + +const readChannelInput = z.object({ + limit: z + .number() + .int() + .min(1) + .max(MAX_CONTEXT_MESSAGES) + .optional() + .describe("How many recent messages to read. Defaults to 50."), +}); + +/** + * The tool and the prompt note that let the agent use what the user is + * currently looking at. + * + * The model is told about the channel but has to ask for it: reading is + * authorized by the user's request, not by them having the channel on screen. + * `<#C…>` renders as the channel's name client-side, which is why no + * `conversations.info` call (and no `channels:read` scope) is needed here. + */ +export function channelContextTooling(args: { + slack: WebClient; + channelId: string; +}): { tools: Record; contextNote: string } { + const { slack, channelId } = args; + + const readChannel = tool({ + description: + "Read the recent messages of the Slack channel the user is currently viewing, so you can summarize the incident discussed there. Only call this when the user's request refers to that conversation.", + inputSchema: readChannelInput, + execute: async ({ limit }) => { + try { + const res = await slack.conversations.history({ + channel: channelId, + limit: limit ?? MAX_CONTEXT_MESSAGES, + }); + const messages = (res.messages ?? []) + .map((m) => ({ + user: m.user ?? m.bot_id ?? "unknown", + text: m.text ?? "", + ts: m.ts ?? "", + })) + .filter((m) => m.text) + // Slack returns newest first; the discussion reads forwards. + .reverse(); + return { channelId, messages }; + } catch (err) { + const error = errorCode(err); + logger.info("slack could not read the context channel", { + error, + channelId, + }); + // Handed back as data so the model can explain it, rather than thrown + // — a channel the bot isn't in is an ordinary outcome here, not a bug. + return { channelId, error }; + } + }, + }); + + return { + tools: { read_slack_channel: readChannel }, + contextNote: ` +Slack context: +- The user is currently looking at the Slack channel <#${channelId}>. +- read_slack_channel reads that channel's recent messages. Call it when the request refers to that conversation ("draft an update for this", "what's happening here?", "summarize this incident"). +- Do NOT call it for unrelated questions, and never repeat its contents unprompted. The user's request is what authorizes reading it. +- If it returns \`error: "not_in_channel"\`, say you aren't in <#${channelId}> and ask them to run \`/invite @openstatus\` there. Do not guess at the incident from anything else.`, + }; +} + +function errorCode(err: unknown): string { + const data = (err as { data?: { error?: unknown } })?.data; + return typeof data?.error === "string" ? data.error : "unknown_error"; +} diff --git a/apps/server/src/routes/slack/commands.ts b/apps/server/src/routes/slack/commands.ts index 1a4e0900..d96a7381 100644 --- a/apps/server/src/routes/slack/commands.ts +++ b/apps/server/src/routes/slack/commands.ts @@ -1,3 +1,4 @@ +import { getLogger } from "@logtape/logtape"; import { ForbiddenError } from "@openstatus/services"; import { createSlackSubscriber, @@ -8,16 +9,22 @@ import { WebClient } from "@slack/web-api"; import type { Context } from "hono"; import { z } from "zod"; +import { runInBackground } from "./background"; import { resolvePageFromUrl } from "./resolve-page"; import { resolveWorkspace } from "./workspace-resolver"; +const logger = getLogger("api-server"); + const slashCommandSchema = z.object({ text: z.string().optional().default(""), team_id: z.string(), channel_id: z.string(), channel_name: z.string().optional(), + response_url: z.string().optional(), }); +type SlashCommand = z.infer; + const HELP = [ "*openstatus*", "• `/openstatus subscribe ` — subscribe this channel to a status page", @@ -29,6 +36,21 @@ function ephemeral(c: Context, text: string) { return c.json({ response_type: "ephemeral", text }); } +/** Deliver a reply after the ack, via the command's single-use response URL. */ +async function respondLater(responseUrl: string, text: string): Promise { + const res = await fetch(responseUrl, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ response_type: "ephemeral", text }), + }); + if (!res.ok) { + logger.error("slack response_url delivery failed", { + status: res.status, + body: await res.text().catch(() => ""), + }); + } +} + async function joinChannel(teamId: string, channelId: string): Promise { try { const resolved = await resolveWorkspace(teamId); @@ -37,117 +59,135 @@ async function joinChannel(teamId: string, channelId: string): Promise { await client.conversations.join({ channel: channelId }); } catch (error) { // Private channels can't be self-joined — the bot must be /invited. - if (error instanceof Error) { - console.error( - `slack: conversations.join failed for ${channelId}: ${error.message}`, - ); - } + logger.warn("slack conversations.join failed", { error, channelId }); } } -export async function handleSlackCommand(c: Context) { +export function handleSlackCommand(c: Context) { const parsed = slashCommandSchema.safeParse(c.get("slackBody")); if (!parsed.success) { return ephemeral(c, "Could not read the command."); } - const { text, team_id, channel_id, channel_name } = parsed.data; + const command = parsed.data; + const sub = subcommand(command); + + // `help` — and anything unrecognised, which falls through to it — needs no + // I/O, so it is answered in the ack itself. + if (sub !== "subscribe" && sub !== "unsubscribe" && sub !== "subscriptions") { + return ephemeral(c, HELP); + } + + // The rest resolve a page, write to the DB and call the Slack API, which can + // outrun the 3s a slash command has to be acknowledged in. Ack now and + // deliver the reply to `response_url` (valid for 30 minutes). + const responseUrl = command.response_url; + if (!responseUrl) { + // Slack always sends one; without it there is nowhere to deliver a late + // reply, so fall back to answering inline. + return runCommand(command).then((text) => ephemeral(c, text)); + } - const tokens = text.trim().split(/\s+/).filter(Boolean); - const sub = (tokens[0] ?? "help").toLowerCase(); - const arg = tokens[1]; + runInBackground( + `command ${sub}`, + async () => { + const text = await runCommand(command); + await respondLater(responseUrl, text); + }, + { teamId: command.team_id, channelId: command.channel_id }, + ); + + return c.body(null, 200); +} + +function subcommand(command: SlashCommand): string { + const tokens = command.text.trim().split(/\s+/).filter(Boolean); + return (tokens[0] ?? "help").toLowerCase(); +} + +function argument(command: SlashCommand): string | undefined { + return command.text.trim().split(/\s+/).filter(Boolean)[1]; +} + +/** Runs the subcommand and returns the message to show the user. */ +async function runCommand(command: SlashCommand): Promise { + const { + team_id: teamId, + channel_id: channelId, + channel_name: channelName, + } = command; + const sub = subcommand(command); + const arg = argument(command); if (sub === "subscribe") { if (!arg) { - return ephemeral(c, "Usage: `/openstatus subscribe `"); + return "Usage: `/openstatus subscribe `"; } const page = await resolvePageFromUrl(arg); if (!page) { - return ephemeral(c, `Couldn't find a status page at \`${arg}\`.`); + return `Couldn't find a status page at \`${arg}\`.`; } try { const result = await createSlackSubscriber({ input: { pageId: page.id, - teamId: team_id, - channelId: channel_id, - channelName: channel_name, + teamId, + channelId, + channelName, }, }); - await joinChannel(team_id, channel_id); + await joinChannel(teamId, channelId); if (result.alreadySubscribed) { - return ephemeral( - c, - `This channel is already subscribed to *${page.title}*.`, - ); + return `This channel is already subscribed to *${page.title}*.`; } - return ephemeral( - c, - `📡 This channel is now subscribed to *${page.title}*. Incident updates will appear here.`, - ); + return `📡 This channel is now subscribed to *${page.title}*. Incident updates will appear here.`; } catch (error) { if (error instanceof ForbiddenError) { - return ephemeral( - c, - `*${page.title}* isn't on a plan that supports subscribers.`, - ); + return `*${page.title}* isn't on a plan that supports subscribers.`; } - console.error("slack /openstatus subscribe failed:", error); - return ephemeral(c, "Something went wrong subscribing this channel."); + logger.error("slack /openstatus subscribe failed", { + error, + teamId, + channelId, + }); + return "Something went wrong subscribing this channel."; } } if (sub === "unsubscribe") { if (!arg) { const subs = await listSlackSubscribersForChannel({ - input: { teamId: team_id, channelId: channel_id }, + input: { teamId, channelId }, }); if (subs.length === 0) { - return ephemeral( - c, - "This channel isn't subscribed to any status page.", - ); + return "This channel isn't subscribed to any status page."; } if (subs.length === 1) { await removeSlackSubscriber({ - input: { - pageId: subs[0].pageId, - teamId: team_id, - channelId: channel_id, - }, + input: { pageId: subs[0].pageId, teamId, channelId }, }); - return ephemeral(c, `Unsubscribed from *${subs[0].pageName}*.`); + return `Unsubscribed from *${subs[0].pageName}*.`; } const list = subs.map((s) => `• ${s.pageName}`).join("\n"); - return ephemeral( - c, - `This channel is subscribed to several pages — specify which:\n${list}\n\nUsage: \`/openstatus unsubscribe \``, - ); + return `This channel is subscribed to several pages — specify which:\n${list}\n\nUsage: \`/openstatus unsubscribe \``; } const page = await resolvePageFromUrl(arg); if (!page) { - return ephemeral(c, `Couldn't find a status page at \`${arg}\`.`); + return `Couldn't find a status page at \`${arg}\`.`; } const { removed } = await removeSlackSubscriber({ - input: { pageId: page.id, teamId: team_id, channelId: channel_id }, + input: { pageId: page.id, teamId, channelId }, }); - return ephemeral( - c, - removed - ? `Unsubscribed from *${page.title}*.` - : `This channel wasn't subscribed to *${page.title}*.`, - ); + return removed + ? `Unsubscribed from *${page.title}*.` + : `This channel wasn't subscribed to *${page.title}*.`; } - if (sub === "subscriptions") { - const subs = await listSlackSubscribersForChannel({ - input: { teamId: team_id, channelId: channel_id }, - }); - if (subs.length === 0) { - return ephemeral(c, "This channel isn't subscribed to any status page."); - } - const list = subs.map((s) => `• *${s.pageName}*`).join("\n"); - return ephemeral(c, `This channel is subscribed to:\n${list}`); + const subs = await listSlackSubscribersForChannel({ + input: { teamId, channelId }, + }); + if (subs.length === 0) { + return "This channel isn't subscribed to any status page."; } - - return ephemeral(c, HELP); + const list = subs.map((s) => `• *${s.pageName}*`).join("\n"); + return `This channel is subscribed to:\n${list}`; } diff --git a/apps/server/src/routes/slack/confirmation-store.test.ts b/apps/server/src/routes/slack/confirmation-store.test.ts index 93e7a211..985f036b 100644 --- a/apps/server/src/routes/slack/confirmation-store.test.ts +++ b/apps/server/src/routes/slack/confirmation-store.test.ts @@ -16,7 +16,7 @@ const redisStore = (globalThis as Record) function makePendingInput(): Omit { return { workspaceId: 1, - botToken: "xoxb-test-token", + teamId: "T_KNOWN", channelId: "C123", threadTs: "1234567890.123456", messageTs: "1234567890.654321", @@ -176,7 +176,7 @@ describe("confirmation-store", () => { const raw = JSON.stringify({ id: "test", workspaceId: 1, - botToken: "tok", + teamId: "T_KNOWN", channelId: "C1", threadTs: "1.1", messageTs: "1.2", diff --git a/apps/server/src/routes/slack/confirmation-store.ts b/apps/server/src/routes/slack/confirmation-store.ts index 1f17f7c7..b15ebd49 100644 --- a/apps/server/src/routes/slack/confirmation-store.ts +++ b/apps/server/src/routes/slack/confirmation-store.ts @@ -14,7 +14,11 @@ export type PendingPayload = z.infer; const pendingActionSchema = z.object({ id: z.string(), workspaceId: z.number(), - botToken: z.string(), + // The workspace to resolve a bot token from when the card is clicked. The + // token itself is deliberately not stored: a pending action outlives the + // turn that made it, and `app_uninstalled` cleans up the integration row, + // not these keys — a stored token would outlive its own install. + teamId: z.string(), channelId: z.string(), threadTs: z.string(), messageTs: z.string(), @@ -39,7 +43,13 @@ export interface CarrierStore { replace(id: string, payload: PendingPayload): Promise; } -const TTL_SECONDS = 5 * 60; +/** + * A storage backstop, not a deadline. Approving is the user's call and a draft + * stays clickable for as long as they need — the card is a Slack message, so + * its age is visible next to it. This only stops abandoned drafts from + * accumulating forever. + */ +const TTL_SECONDS = 30 * 24 * 60 * 60; const ACTION_PREFIX = "slack:action:"; const THREAD_PREFIX = "slack:thread:"; diff --git a/apps/server/src/routes/slack/handler.test.ts b/apps/server/src/routes/slack/handler.test.ts index a8eb1801..9a198883 100644 --- a/apps/server/src/routes/slack/handler.test.ts +++ b/apps/server/src/routes/slack/handler.test.ts @@ -16,7 +16,9 @@ import { handleSlackEvent, isAnswerToAgent, looksLikeUncardedDraft, + toolTaskTitle, } from "./handler"; +import { abortTurn, endTurn, startTurn } from "./running-turns"; import { verifySlackSignature } from "./verify"; function createTestApp() { @@ -48,45 +50,54 @@ function signAndPost( }); } +function resetSlackTestState() { + slackTestState.calls = []; + slackTestState.postMessageOverride = null; + slackTestState.updateOverride = null; + slackTestState.runAgentOverride = null; + slackTestState.renameOverride = null; + slackTestState.chatStreamEnabled = true; + slackTestState.historyImpl = () => + Promise.resolve({ + messages: [{ user: "U1", text: "channel message", ts: "1.1" }], + }); + slackTestState.streamAppendFailAfter = null; + slackTestState.repliesImpl = () => + Promise.resolve({ + messages: [{ user: "U1", text: "test message", ts: "1.1" }], + }); + // Default to a workspace without agent sessions so the tests below cover + // the fallback indicators; the session path opts back in explicitly. + slackTestState.sessionStatusOverride = () => { + const err = new Error("An API error occurred: feature_disabled"); + Object.assign(err, { + code: "slack_webapi_platform_error", + data: { ok: false, error: "feature_disabled" }, + }); + return Promise.reject(err); + }; + slackTestState.resolveWorkspace = (teamId: string) => { + if (teamId === "T_KNOWN") { + return Promise.resolve({ + workspace: { + id: 1, + name: "Test Workspace", + slug: "test", + plan: "free", + limits: {}, + }, + botToken: "xoxb-test", + botUserId: "UBOT", + }); + } + return Promise.resolve(null); + }; +} + describe("handleSlackEvent", () => { const app = createTestApp(); - beforeEach(() => { - slackTestState.calls = []; - slackTestState.postMessageOverride = null; - slackTestState.updateOverride = null; - slackTestState.runAgentOverride = null; - slackTestState.repliesImpl = () => - Promise.resolve({ - messages: [{ user: "U1", text: "test message", ts: "1.1" }], - }); - // Default to a workspace without agent sessions so the tests below cover - // the fallback indicators; the session path opts back in explicitly. - slackTestState.sessionStatusOverride = () => { - const err = new Error("An API error occurred: feature_disabled"); - Object.assign(err, { - code: "slack_webapi_platform_error", - data: { ok: false, error: "feature_disabled" }, - }); - return Promise.reject(err); - }; - slackTestState.resolveWorkspace = (teamId: string) => { - if (teamId === "T_KNOWN") { - return Promise.resolve({ - workspace: { - id: 1, - name: "Test Workspace", - slug: "test", - plan: "free", - limits: {}, - }, - botToken: "xoxb-test", - botUserId: "UBOT", - }); - } - return Promise.resolve(null); - }; - }); + beforeEach(resetSlackTestState); test("responds to url_verification challenge", async () => { const res = await signAndPost(app, { @@ -542,7 +553,7 @@ describe("handleSlackEvent", () => { expect(errorPost).toBeDefined(); }); - test("sets suggested prompts on assistant_thread_started", async () => { + test("greets in the thread on the legacy assistant_thread_started", async () => { const res = await signAndPost(app, { type: "event_callback", team_id: "T_KNOWN", @@ -561,12 +572,14 @@ describe("handleSlackEvent", () => { expect(res.status).toBe(200); await new Promise((r) => setTimeout(r, 50)); - const prompts = slackTestState.calls.find( - (m) => m.method === "assistant.threads.setSuggestedPrompts", + // The greeting lands in the thread that was just opened, and states the + // approval guarantee where it matters rather than only in App Home. + const welcome = slackTestState.calls.find( + (m) => m.method === "postMessage", ); - expect(prompts?.args).toMatchObject({ channel_id: "D1", thread_ts: "2.2" }); - expect(prompts?.args.prompts).toBeDefined(); - expect((prompts?.args.prompts as unknown[]).length).toBeGreaterThan(0); + expect(welcome?.args).toMatchObject({ channel: "D1", thread_ts: "2.2" }); + expect(welcome?.args.text as string).toContain("Approve"); + expect((welcome?.args.blocks as unknown[]).length).toBeGreaterThan(0); }); test("ignores events without channel", async () => { @@ -780,13 +793,15 @@ describe("handleSlackEvent", () => { expect(res.status).toBe(200); await new Promise((r) => setTimeout(r, 100)); - const errorUpdate = slackTestState.calls.find( + // Delivered as a fresh message when streaming, or by overwriting the + // "Thinking..." placeholder when it isn't — either way the user sees it. + const errorMessage = slackTestState.calls.find( (m) => - m.method === "update" && + (m.method === "update" || m.method === "postMessage") && typeof m.args.text === "string" && m.args.text.includes("Something went wrong"), ); - expect(errorUpdate).toBeDefined(); + expect(errorMessage).toBeDefined(); }); test("does not throw when both runAgent and error update fail", async () => { @@ -916,3 +931,613 @@ describe("isAnswerToAgent", () => { ); }); }); + +describe("streaming the agent's answer", () => { + const app = createTestApp(); + + beforeEach(resetSlackTestState); + + /** Drives the agent mock's `events` so the handler sees a real stream. */ + function streamTurn( + drive: (events: { + onTextDelta(delta: string): Promise; + onToolCall(c: { id: string; toolName: string }): Promise; + onToolResult(r: { id: string; toolName: string }): Promise; + }) => Promise, + text = "All five monitors are healthy.", + ) { + slackTestState.runAgentOverride = async (options: unknown) => { + // biome-ignore lint/suspicious/noExplicitAny: test double plumbing + await drive((options as any).events); + return { + text, + toolResults: [], + finishReason: "stop", + stepCount: 1, + hitStepLimit: false, + aborted: false, + }; + }; + } + + function mention(suffix: string) { + return signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: `evt_stream_${suffix}`, + event: { + type: "app_mention", + text: "<@UBOT> what's broken?", + user: "U1", + channel: "C1", + ts: `${Date.now()}.${suffix}`, + }, + }); + } + + test("streams the answer instead of posting it", async () => { + streamTurn(async (events) => { + await events.onTextDelta("All five monitors "); + await events.onTextDelta("are healthy."); + }); + + const res = await mention("1"); + expect(res.status).toBe(200); + await new Promise((r) => setTimeout(r, 100)); + + const appended = slackTestState.calls + .filter((m) => m.method === "stream.append") + .map((m) => m.args.markdown_text); + expect(appended).toEqual(["All five monitors ", "are healthy."]); + + // The stream carried the answer, so it is finalized rather than re-posted. + expect(slackTestState.calls.some((m) => m.method === "stream.stop")).toBe( + true, + ); + expect(slackTestState.calls.some((m) => m.method === "postMessage")).toBe( + false, + ); + }); + + test("reports each tool call as a task", async () => { + streamTurn(async (events) => { + await events.onToolCall({ id: "t1", toolName: "list_status_pages" }); + await events.onToolResult({ id: "t1", toolName: "list_status_pages" }); + await events.onTextDelta("Done."); + }); + + await mention("2"); + await new Promise((r) => setTimeout(r, 100)); + + const tasks = slackTestState.calls + .filter((m) => m.method === "stream.append" && m.args.chunks) + .flatMap((m) => m.args.chunks as Record[]); + expect(tasks).toEqual([ + { + type: "task_update", + id: "t1", + title: "Reading status pages", + status: "in_progress", + }, + { + type: "task_update", + id: "t1", + title: "Reading status pages", + status: "complete", + }, + ]); + }); + + test("falls back to the placeholder when the workspace has no streaming", async () => { + slackTestState.chatStreamEnabled = false; + streamTurn(async () => {}); + + await mention("3"); + await new Promise((r) => setTimeout(r, 100)); + + // No agent session and no stream leaves the oldest path: a "Thinking..." + // message posted up front and overwritten with the answer. + const placeholder = slackTestState.calls.find( + (m) => m.method === "postMessage", + ); + expect(placeholder?.args.text).toContain("Thinking..."); + const answer = slackTestState.calls.find((m) => m.method === "update"); + expect(answer?.args.text).toBe("All five monitors are healthy."); + expect(slackTestState.calls.some((m) => m.method === "chatStream")).toBe( + false, + ); + }); + + test("rewrites the partial message when the stream breaks mid-turn", async () => { + // The first append lands, the second fails — Slack is left holding half + // an answer, so the whole answer has to replace it. + slackTestState.streamAppendFailAfter = 1; + streamTurn(async (events) => { + await events.onTextDelta("All five "); + await events.onTextDelta("monitors are healthy."); + }); + + await mention("4"); + await new Promise((r) => setTimeout(r, 100)); + + const rewrite = slackTestState.calls.find((m) => m.method === "update"); + expect(rewrite?.args).toMatchObject({ + channel: "C1", + ts: "stream.ts", + text: "All five monitors are healthy.", + }); + // Nothing is posted alongside it: one message, one answer. + expect(slackTestState.calls.some((m) => m.method === "postMessage")).toBe( + false, + ); + }); + + test("names tasks after the tool's verb", () => { + expect(toolTaskTitle("list_status_pages")).toBe("Reading status pages"); + expect(toolTaskTitle("get_monitor_status")).toBe("Reading monitor status"); + expect(toolTaskTitle("create_status_report")).toBe( + "Drafting status report", + ); + expect(toolTaskTitle("search_docs")).toBe("Searching docs"); + // An unknown verb still reads as words rather than a tool name. + expect(toolTaskTitle("frobnicate_widgets")).toBe("frobnicate widgets"); + expect(toolTaskTitle("ping")).toBe("ping"); + }); +}); + +describe("running turns", () => { + test("aborts only the thread it was asked about", () => { + const turn = startTurn("C_RT", "1.1"); + expect(abortTurn("C_RT", "9.9")).toBe(false); + expect(turn.signal.aborted).toBe(false); + + expect(abortTurn("C_RT", "1.1")).toBe(true); + expect(turn.signal.aborted).toBe(true); + + endTurn("C_RT", "1.1", turn); + expect(abortTurn("C_RT", "1.1")).toBe(false); + }); + + test("a finished turn does not deregister the one that replaced it", () => { + const first = startTurn("C_RT2", "2.2"); + const second = startTurn("C_RT2", "2.2"); + endTurn("C_RT2", "2.2", first); + + expect(abortTurn("C_RT2", "2.2")).toBe(true); + expect(second.signal.aborted).toBe(true); + endTurn("C_RT2", "2.2", second); + }); +}); + +describe("stopping a turn", () => { + const app = createTestApp(); + + beforeEach(() => { + resetSlackTestState(); + // A workspace with agent sessions — the surface the stop button lives on. + slackTestState.sessionStatusOverride = null; + }); + + function stopEvent(channel: string, threadTs: string) { + return signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: `evt_stop_${channel}_${threadTs}`, + event: { + type: "agent_session_stopped", + channel, + thread_ts: threadTs, + user: "U1", + streaming_message_ts: [], + event_ts: "1.1", + }, + }); + } + + function activeStatusCalls() { + return slackTestState.calls.filter( + (m) => + m.method === "agents.sessions.setStatus" && m.args.status === "active", + ); + } + + test("aborts the run, confirms the stop, and clears the status", async () => { + let signal: AbortSignal | undefined; + slackTestState.runAgentOverride = (options: unknown) => { + signal = (options as { signal: AbortSignal }).signal; + return new Promise((resolve) => { + signal?.addEventListener("abort", () => + resolve({ + text: "Looking at the API monit", + toolResults: [], + finishReason: "abort", + stepCount: 0, + hitStepLimit: false, + aborted: true, + }), + ); + }); + }; + + const ts = "6001.1"; + await signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: `evt_stopme_${ts}`, + event: { + type: "app_mention", + text: "<@UBOT> what's broken?", + user: "U1", + channel: "C_STOP", + ts, + }, + }); + await new Promise((r) => setTimeout(r, 50)); + expect(signal).toBeDefined(); + expect(signal?.aborted).toBe(false); + + await stopEvent("C_STOP", ts); + await new Promise((r) => setTimeout(r, 100)); + + expect(signal?.aborted).toBe(true); + expect(activeStatusCalls().length).toBeGreaterThan(0); + + const notice = slackTestState.calls.find( + (m) => + m.method === "postMessage" && + typeof m.args.text === "string" && + m.args.text.includes("Stopped"), + ); + expect(notice).toBeDefined(); + + // The half-written answer is never delivered as if it were finished. + const answer = slackTestState.calls.find( + (m) => + typeof m.args.text === "string" && + m.args.text.includes("Looking at the API monit"), + ); + expect(answer).toBeUndefined(); + }); + + test("clears the status even when no turn is running here", async () => { + await stopEvent("C_STOP2", "7001.1"); + await new Promise((r) => setTimeout(r, 50)); + + // Nothing to abort — another instance may hold the turn — but the user + // still has to get out of the loading state. + expect(activeStatusCalls()).toHaveLength(1); + expect(activeStatusCalls()[0].args).toMatchObject({ + channel_id: "C_STOP2", + thread_ts: "7001.1", + }); + }); +}); + +describe("titling a thread", () => { + const app = createTestApp(); + const redisStore = (globalThis as Record) + .__testRedisStore as Map; + + beforeEach(() => { + resetSlackTestState(); + redisStore.clear(); + }); + + function renameCalls() { + return slackTestState.calls.filter( + (m) => m.method === "agents.sessions.rename", + ); + } + + function paneMessage(ts: string, text: string, threadTs?: string) { + return signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: `evt_title_${ts}`, + event: { + type: "message", + channel_type: "im", + text, + user: "U1", + channel: "D_TITLE", + ts, + ...(threadTs ? { thread_ts: threadTs } : {}), + }, + }); + } + + test("names the pane thread after what the user asked", async () => { + await paneMessage("8001.1", "which reports are currently open?"); + await new Promise((r) => setTimeout(r, 100)); + + expect(renameCalls()).toHaveLength(1); + expect(renameCalls()[0].args).toMatchObject({ + channel_id: "D_TITLE", + thread_ts: "8001.1", + title: "which reports are currently open?", + }); + }); + + test("names it once and leaves it alone after that", async () => { + await paneMessage("8002.1", "is the checkout monitor healthy?"); + await new Promise((r) => setTimeout(r, 100)); + expect(renameCalls()).toHaveLength(1); + + // A second turn on the same thread: the subject hasn't changed, and the + // name shouldn't follow whatever was asked next. + await paneMessage("8002.2", "and what about billing?", "8002.1"); + await new Promise((r) => setTimeout(r, 100)); + expect(renameCalls()).toHaveLength(1); + }); + + test("stops renaming once a person has named it", async () => { + await signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: "evt_renamed_8003", + event: { + type: "agent_session_title_changed", + channel: "D_TITLE", + thread_ts: "8003.1", + user: "U1", + title: "Tuesday's Stripe outage", + event_ts: "1.1", + }, + }); + await new Promise((r) => setTimeout(r, 50)); + + await paneMessage("8003.2", "any update on this?", "8003.1"); + await new Promise((r) => setTimeout(r, 100)); + + expect(renameCalls()).toHaveLength(0); + }); + + test("leaves channel threads alone", async () => { + // `agents.sessions.rename` also renames the channel for session channels — + // not worth risking on a shared incident channel for a name nobody lists. + await signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: "evt_title_channel", + event: { + type: "app_mention", + text: "<@UBOT> what's broken?", + user: "U1", + channel: "C_TITLE", + ts: "8004.1", + }, + }); + await new Promise((r) => setTimeout(r, 100)); + + expect(renameCalls()).toHaveLength(0); + }); + + test("keeps the answer when renaming fails", async () => { + slackTestState.renameOverride = () => + Promise.reject(new Error("feature_disabled")); + + await paneMessage("8005.1", "which reports are open?"); + await new Promise((r) => setTimeout(r, 100)); + + const answered = slackTestState.calls.some( + (m) => + typeof m.args.text === "string" && + m.args.text.includes("Here is my response"), + ); + expect(answered).toBe(true); + // A failed rename must not mark the thread as named. + expect(redisStore.has("slack:title:D_TITLE:8005.1")).toBe(false); + }); +}); + +describe("greeting on first contact", () => { + const app = createTestApp(); + const redisStore = (globalThis as Record) + .__testRedisStore as Map; + + beforeEach(() => { + resetSlackTestState(); + redisStore.clear(); + }); + + function homeOpened(tab: string, userId = "U1") { + return signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: `evt_home_${tab}_${userId}_${Math.random()}`, + event: { + type: "app_home_opened", + user: userId, + channel: "D_WELCOME", + tab, + event_ts: "1.1", + }, + }); + } + + function welcomes() { + return slackTestState.calls.filter( + (m) => + m.method === "postMessage" && + typeof m.args.text === "string" && + m.args.text.includes("Approve"), + ); + } + + test("greets when the Messages tab is opened", async () => { + await homeOpened("messages"); + await new Promise((r) => setTimeout(r, 50)); + + expect(welcomes()).toHaveLength(1); + // Top-level in the DM: the agent experience has no thread to greet into. + expect(welcomes()[0].args.thread_ts).toBeUndefined(); + }); + + test("greets a person once, however often they open it", async () => { + await homeOpened("messages"); + await new Promise((r) => setTimeout(r, 50)); + await homeOpened("messages"); + await new Promise((r) => setTimeout(r, 50)); + + expect(welcomes()).toHaveLength(1); + }); + + test("greets each person separately", async () => { + await homeOpened("messages", "U1"); + await new Promise((r) => setTimeout(r, 50)); + await homeOpened("messages", "U2"); + await new Promise((r) => setTimeout(r, 50)); + + expect(welcomes()).toHaveLength(2); + }); + + test("publishes the home view on the Home tab without greeting", async () => { + await homeOpened("home"); + await new Promise((r) => setTimeout(r, 50)); + + expect(slackTestState.calls.some((m) => m.method === "views.publish")).toBe( + true, + ); + expect(welcomes()).toHaveLength(0); + }); + + test("does not mark someone greeted when the greeting fails", async () => { + slackTestState.postMessageOverride = () => + Promise.reject(new Error("channel_not_found")); + + await homeOpened("messages"); + await new Promise((r) => setTimeout(r, 50)); + + // Otherwise a transient failure costs them the greeting permanently. + expect(redisStore.has("slack:greeted:T_KNOWN:U1")).toBe(false); + }); +}); + +describe("the channel the user is viewing", () => { + const app = createTestApp(); + const redisStore = (globalThis as Record) + .__testRedisStore as Map; + + beforeEach(() => { + resetSlackTestState(); + redisStore.clear(); + }); + + function contextChanged( + entities: Array> | undefined, + userId = "U1", + ) { + return signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: `evt_ctx_${userId}_${Math.random()}`, + event: { + type: "app_context_changed", + context: entities ? { entities } : {}, + }, + authorizations: [ + { user_id: "B0", is_bot: true }, + { user_id: userId, is_bot: false }, + ], + }); + } + + /** Captures what the handler handed the agent for this turn. */ + function captureAgentOptions() { + const seen: { tools?: Record; contextNote?: string }[] = + []; + slackTestState.runAgentOverride = (options: unknown) => { + seen.push(options as { contextNote?: string }); + return Promise.resolve({ + text: "Here is my response", + toolResults: [], + finishReason: "stop", + stepCount: 1, + hitStepLimit: false, + aborted: false, + }); + }; + return seen; + } + + test("remembers it for the authorizing human", async () => { + await contextChanged([ + { type: "slack#/types/channel_id", value: "C_INCIDENT" }, + ]); + await new Promise((r) => setTimeout(r, 50)); + + expect(redisStore.get("slack:context:T_KNOWN:U1")).toBe("C_INCIDENT"); + // Never attributed to the bot authorization. + expect(redisStore.has("slack:context:T_KNOWN:B0")).toBe(false); + }); + + test("forgets it when the context empties", async () => { + await contextChanged([ + { type: "slack#/types/channel_id", value: "C_INCIDENT" }, + ]); + await new Promise((r) => setTimeout(r, 50)); + await contextChanged(undefined); + await new Promise((r) => setTimeout(r, 50)); + + // A channel they left is worse than no context at all. + expect(redisStore.has("slack:context:T_KNOWN:U1")).toBe(false); + }); + + test("offers the channel to the agent on a pane turn", async () => { + await contextChanged([ + { type: "slack#/types/channel_id", value: "C_INCIDENT" }, + ]); + await new Promise((r) => setTimeout(r, 50)); + + const seen = captureAgentOptions(); + await signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: "evt_ctx_turn", + event: { + type: "message", + channel_type: "im", + text: "draft an update for this", + user: "U1", + channel: "D_CTX", + ts: "9101.1", + }, + }); + await new Promise((r) => setTimeout(r, 100)); + + expect(seen).toHaveLength(1); + expect(seen[0].contextNote).toContain("<#C_INCIDENT>"); + expect(Object.keys(seen[0].tools ?? {})).toContain("read_slack_channel"); + // Nothing is read until the model decides the request calls for it. + expect( + slackTestState.calls.some((m) => m.method === "conversations.history"), + ).toBe(false); + }); + + test("leaves channel turns alone", async () => { + await contextChanged([ + { type: "slack#/types/channel_id", value: "C_INCIDENT" }, + ]); + await new Promise((r) => setTimeout(r, 50)); + + const seen = captureAgentOptions(); + await signAndPost(app, { + type: "event_callback", + team_id: "T_KNOWN", + event_id: "evt_ctx_channel_turn", + event: { + type: "app_mention", + text: "<@UBOT> what's broken?", + user: "U1", + channel: "C_OTHER", + ts: "9102.1", + }, + }); + await new Promise((r) => setTimeout(r, 100)); + + // In a channel the agent already has the thread it was called into. + expect(seen).toHaveLength(1); + expect(seen[0].contextNote).toBeUndefined(); + expect(seen[0].tools).toBeUndefined(); + }); +}); diff --git a/apps/server/src/routes/slack/handler.ts b/apps/server/src/routes/slack/handler.ts index af266dc0..2dbcd463 100644 --- a/apps/server/src/routes/slack/handler.ts +++ b/apps/server/src/routes/slack/handler.ts @@ -5,28 +5,39 @@ import { WebClient } from "@slack/web-api"; import type { Context } from "hono"; import { z } from "zod"; -import { runAgent } from "./agent"; -import { - setAssistantStatus, - setSessionStatus, - startAssistantThread, -} from "./assistant"; +import { type AgentEvents, runAgent } from "./agent"; +import { greetOnce, setAssistantStatus, setSessionStatus } from "./assistant"; import { type Block, + buildAnswerMessage, buildConfirmationBlocks, getConfirmationText, type RefResolvers, } from "./blocks"; +import { + channelContextTooling, + contextChannelId, + contextUserId, + forgetContext, + recallContext, + rememberContext, +} from "./channel-context"; import { findByThread, replace, store } from "./confirmation-store"; import type { PendingPayload } from "./confirmation-store"; import { publishHomeView } from "./home"; -import { toMrkdwn } from "./mrkdwn"; import { getComponentNames, getPageDashboardLink, getStatusReportLink, } from "./page-urls"; import { getRegistryTool, isSlackToolDraft } from "./registry-runner"; +import { abortTurn, endTurn, startTurn } from "./running-turns"; +import { + buildThreadTitle, + isThreadTitled, + markThreadTitled, + renameThread, +} from "./thread-title"; import { resolveWorkspace } from "./workspace-resolver"; function makeRefResolvers(workspaceId: number): RefResolvers { @@ -70,6 +81,21 @@ const slackEventSchema = z.object({ .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(), }) @@ -77,6 +103,15 @@ const slackEventSchema = z.object({ 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; @@ -229,35 +264,61 @@ async function processEvent(body: SlackEvent) { 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") { - if (event.tab && event.tab !== "home") return; + 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); + + if (tab === "messages") { + if (!event.channel) return; + try { + await greetOnce({ + slack, + teamId, + userId, + channel: event.channel, + }); + } catch (err) { + logger.error("slack failed to greet user", { error: err, teamId }); + } + return; + } + try { - await publishHomeView(new WebClient(resolved.botToken), userId); + await publishHomeView(slack, userId); } 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) return; + if (!teamId || !thread?.user_id) return; const resolved = await resolveWorkspace(teamId); if (!resolved) return; try { - await startAssistantThread( - new WebClient(resolved.botToken), - thread.channel_id, - thread.thread_ts, - ); + await greetOnce({ + slack: new WebClient(resolved.botToken), + teamId, + userId: thread.user_id, + channel: thread.channel_id, + threadTs: thread.thread_ts, + }); } catch (err) { - logger.error("slack failed to start assistant thread", { + logger.error("slack failed to greet in new thread", { error: err, teamId, }); @@ -265,6 +326,70 @@ async function processEvent(body: SlackEvent) { 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 (!event.channel || !event.thread_ts) return; + await markThreadTitled(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); + 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; @@ -346,19 +471,18 @@ async function processEvent(body: SlackEvent) { agentThread: isAgentThread, }); - const reply = - (await acknowledgeWithSession( - slack, - event.channel, - threadTs, - teamId, - event.user, - )) ?? - (isAgentThread - ? await acknowledgeInAgentThread(slack, event.channel, threadTs, teamId) - : await acknowledgeInChannel(slack, event.channel, threadTs, teamId)); + const reply = await createReply({ + slack, + channel: event.channel, + threadTs, + teamId, + userId: event.user, + isAgentThread, + }); if (!reply) return; + const turn = startTurn(event.channel, threadTs); + try { let thread: ThreadMessage[] = []; if (prefetchedThread) { @@ -371,11 +495,22 @@ async function processEvent(body: SlackEvent) { 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; + logger.info("slack agent invoked", { teamId, channel: event.channel, threadTs, messageCount: thread.length, + contextChannel, }); const result = await runAgent( @@ -384,8 +519,24 @@ async function processEvent(body: SlackEvent) { botUserId, event.text, { slackUserId: event.user ?? "", teamId }, + { + events: reply.progress, + signal: turn.signal, + tools: context?.tools, + contextNote: context?.contextNote, + }, ); + 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, @@ -419,7 +570,7 @@ async function processEvent(body: SlackEvent) { threadTs, event.user ?? "", resolved.workspace.id, - resolved.botToken, + teamId, confirmationResult, ); } else { @@ -437,15 +588,31 @@ async function processEvent(body: SlackEvent) { readToolCalls: result.toolResults.map((tr) => tr.toolName), }); } - await reply.send({ - text: result.text ? toMrkdwn(result.text) : "Done!", - }); + 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: + confirmationResult && isSlackToolDraft(confirmationResult.result) + ? confirmationResult.result + : undefined, + userText: event.text, + }); } catch (err) { logger.error("slack agent error", { error: err, @@ -463,38 +630,376 @@ async function processEvent(body: SlackEvent) { }); }); } 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; +} + /** - * Where the agent's answer goes. In a channel we post a "Thinking..." message - * up front and overwrite it; in the agent pane Slack renders a native status - * instead, so the answer is a fresh message in the thread. + * 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 { + const { + slack, + channel, + threadTs, + teamId, + workspaceId, + isAgentThread, + draft, + userText, + } = args; + if (!isAgentThread) return; + + try { + if (await isThreadTitled(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; a native "working" status on the thread; or + * a "Thinking..." message we post up front and overwrite. */ interface Reply { - /** Writes the answer and returns the ts of the message holding it. */ + /** + * 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; + /** Writes a message of our own — a confirmation card, or an error. */ send(message: { text: string; blocks?: Block[] }): Promise; + /** + * Confirms the turn ended because the user stopped it, leaving whatever was + * already written in place. + */ + stopped(): Promise; + /** Progress to report while the turn runs, when the surface can show it. */ + progress?: AgentEvents; /** Our own "Thinking..." message, to keep it out of the agent's context. */ placeholderTs?: string; /** Runs once the turn is over, whether it succeeded or not. */ finish?: () => Promise; } +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 = { + 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>; + channel: string; + threadTs: string; + teamId: string; + finishSession?: () => Promise; +}): 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) => { + 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. */ + const closeStream = async (): Promise => { + if (streamClosed || broken || !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 || broken || !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 { + const { slack, channel, threadTs, teamId, userId, isAgentThread } = args; + + const session = await acknowledgeWithSession( + slack, + channel, + threadTs, + teamId, + userId, + ); + + // The status is the loading indicator, not the delivery: a streamed turn in + // the agent pane still needs it when agent sessions aren't available. + if (!session && isAgentThread) { + // 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, + }); + } + + if (session) return session; + if (isAgentThread) return acknowledgeInAgentThread(slack, channel, threadTs); + return acknowledgeInChannel(slack, channel, threadTs, teamId); +} + function postInThread( slack: WebClient, channel: string, threadTs: string, ): Reply["send"] { return async ({ text, blocks }) => { - 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; + 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; + } }; } @@ -522,8 +1027,13 @@ async function acknowledgeWithSession( }); return; } + const send = postInThread(slack, channel, threadTs); return { - send: postInThread(slack, channel, threadTs), + send, + answer: (text) => send(buildAnswerMessage(text)), + stopped: async () => { + await send({ text: STOPPED_NOTICE }); + }, async finish() { await setSessionStatus(slack, channel, threadTs, "active").catch( (err: unknown) => @@ -537,22 +1047,21 @@ async function acknowledgeWithSession( }; } -async function acknowledgeInAgentThread( +// 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 acknowledgeInAgentThread( slack: WebClient, channel: string, threadTs: string, - teamId: string, -): Promise { - // 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, - }), - ); - return { send: postInThread(slack, channel, threadTs) }; +): Reply { + const send = postInThread(slack, channel, threadTs); + return { + send, + answer: (text) => send(buildAnswerMessage(text)), + stopped: async () => { + await send({ text: STOPPED_NOTICE }); + }, + }; } async function acknowledgeInChannel( @@ -607,11 +1116,17 @@ async function acknowledgeInChannel( } const ts = thinkingTs; + const send: Reply["send"] = async ({ text, blocks }) => { + await slack.chat.update({ channel, ts, text, blocks }); + return ts; + }; return { placeholderTs: ts, - async send({ text, blocks }) { - await slack.chat.update({ channel, ts, text, blocks }); - return ts; + send, + answer: (text) => send(buildAnswerMessage(text)), + // Overwrites "Thinking...", which would otherwise stand forever. + stopped: async () => { + await send({ text: STOPPED_NOTICE }); }, }; } @@ -623,7 +1138,7 @@ async function handleConfirmation( threadTs: string, userId: string, workspaceId: number, - botToken: string, + teamId: string, confirmationResult: { toolName: string; result: unknown }, ) { if (!isSlackToolDraft(confirmationResult.result)) return; @@ -673,7 +1188,7 @@ async function handleConfirmation( const messageTs = await reply.send({ text }); const actionId = await store({ workspaceId, - botToken, + teamId, channelId: channel, threadTs, messageTs, diff --git a/apps/server/src/routes/slack/home.ts b/apps/server/src/routes/slack/home.ts index 8caad375..39a20591 100644 --- a/apps/server/src/routes/slack/home.ts +++ b/apps/server/src/routes/slack/home.ts @@ -1,7 +1,7 @@ import type { WebClient } from "@slack/web-api"; import type { KnownBlock } from "@slack/web-api"; -const DOCS_URL = "https://www.openstatus.dev/docs"; +export const DOCS_URL = "https://www.openstatus.dev/docs"; export function buildHomeBlocks(): KnownBlock[] { return [ diff --git a/apps/server/src/routes/slack/interactions.test.ts b/apps/server/src/routes/slack/interactions.test.ts index 401c77d1..b4f7c65b 100644 --- a/apps/server/src/routes/slack/interactions.test.ts +++ b/apps/server/src/routes/slack/interactions.test.ts @@ -11,6 +11,7 @@ import { withSlackConfig, } from "@/libs/test/slack-config"; +import { settleBackgroundTasks } from "./background"; import type { SlackEnv } from "./config"; import { handleSlackInteraction } from "./interactions"; import { verifySlackSignature } from "./verify"; @@ -21,7 +22,7 @@ const redisStore = (globalThis as Record) const basePending = { id: "pending-123", workspaceId: 1, - botToken: "xoxb-test", + teamId: "T_KNOWN", channelId: "C1", threadTs: "1.1", messageTs: "1.2", @@ -91,7 +92,9 @@ function seedCreateMaintenance( return data; } -function signAndPost( +// The route acks Slack immediately and finishes the work in the background, +// so every test waits for that work before asserting on its side effects. +async function signAndPost( app: ReturnType, payload: Record, ) { @@ -104,7 +107,7 @@ function signAndPost( .update(basestring) .digest("hex"); - return app.request("/slack/interactions", { + const res = await app.request("/slack/interactions", { method: "POST", headers: { "Content-Type": "application/x-www-form-urlencoded", @@ -113,6 +116,8 @@ function signAndPost( }, body, }); + await settleBackgroundTasks(); + return res; } describe("handleSlackInteraction (dispatch)", () => { @@ -224,6 +229,49 @@ describe("handleSlackInteraction (dispatch)", () => { expect(res.status).toBe(200); }); + test("resolves the bot token on click, not from the stored action", async () => { + seedCreateStatusReport(); + let resolveCalls = 0; + slackTestState.resolveWorkspace = (teamId: string) => { + resolveCalls++; + return teamId === "T_KNOWN" + ? Promise.resolve({ botToken: "xoxb-fresh" }) + : Promise.resolve(null); + }; + + await signAndPost(app, { + type: "block_actions", + user: { id: "U_OWNER" }, + channel: { id: "C1" }, + message: { ts: "1.2" }, + team: { id: "T_KNOWN" }, + actions: [{ action_id: "cancel_pending-123" }], + }); + + // A card outlives the turn that made it, so the token that made it may + // have been revoked by now — it is never persisted, only resolved here. + expect(resolveCalls).toBe(1); + expect(slackTestState.calls.some((c) => c.method === "update")).toBe(true); + }); + + test("does nothing when the workspace no longer resolves", async () => { + seedCreateStatusReport(); + slackTestState.resolveWorkspace = () => Promise.resolve(null); + + await signAndPost(app, { + type: "block_actions", + user: { id: "U_OWNER" }, + channel: { id: "C1" }, + message: { ts: "1.2" }, + team: { id: "T_KNOWN" }, + actions: [{ action_id: "approve_pending-123" }], + }); + + expect(slackTestState.calls).toHaveLength(0); + // The action survives an uninstall rather than being silently burned. + expect(redisStore.has("slack:action:pending-123")).toBe(true); + }); + test("cancel consumes pending from redis", async () => { seedCreateStatusReport(); await signAndPost(app, { diff --git a/apps/server/src/routes/slack/interactions.ts b/apps/server/src/routes/slack/interactions.ts index 7503cd3d..4e0a569b 100644 --- a/apps/server/src/routes/slack/interactions.ts +++ b/apps/server/src/routes/slack/interactions.ts @@ -1,8 +1,10 @@ +import { getLogger } from "@logtape/logtape"; import { ServiceError } from "@openstatus/services"; import { WebClient } from "@slack/web-api"; import type { Context } from "hono"; -import { parseActionId } from "./blocks"; +import { runInBackground } from "./background"; +import { type ParsedActionId, parseActionId } from "./blocks"; import { consume, get } from "./confirmation-store"; import type { PendingAction } from "./confirmation-store"; import { renderToolResult } from "./presenters"; @@ -10,6 +12,8 @@ import { executeRegistryAction, getRegistryTool } from "./registry-runner"; import { toServiceCtx } from "./service-adapter"; import { resolveWorkspace } from "./workspace-resolver"; +const logger = getLogger("api-server"); + interface SlackInteractionPayload { type: string; user: { id: string }; @@ -19,7 +23,7 @@ interface SlackInteractionPayload { actions: Array<{ action_id: string; value?: string }>; } -export async function handleSlackInteraction(c: Context) { +export function handleSlackInteraction(c: Context) { const payload = c.get("slackBody") as SlackInteractionPayload; if (payload.type !== "block_actions" || !payload.actions?.length) { @@ -29,22 +33,39 @@ export async function handleSlackInteraction(c: Context) { const parsed = parseActionId(payload.actions[0].action_id); if (!parsed) return c.json({ ok: true }); + // Executing the action writes to the DB and can notify every subscriber of + // the status page — well past Slack's 3s ack window, which would mark the + // click as failed even though it worked. Ack now; the card is updated with + // the outcome when the work finishes. + runInBackground("interaction", () => processInteraction(parsed, payload), { + actionId: payload.actions[0].action_id, + teamId: payload.team?.id, + }); + + return c.json({ ok: true }); +} + +async function processInteraction( + parsed: ParsedActionId, + payload: SlackInteractionPayload, +) { const channelId = payload.channel.id; const messageTs = payload.message.ts; const userId = payload.user.id; const teamId = payload.team?.id; - // Non-atomic read for botToken resolution and authorization checks + // Non-atomic read, for the authorization checks below. const pending = await get(parsed.pendingId); - let botToken: string | undefined = pending?.botToken; - if (!botToken && teamId) { - const resolved = await resolveWorkspace(teamId); - botToken = resolved?.botToken; - } - if (!botToken) return c.json({ ok: true }); + // Resolved at click time, never stored: the token that made the card may + // have been revoked since. The payload's team is authoritative for who + // clicked; the pending's is the fallback when Slack omits it. + const workspaceTeamId = teamId ?? pending?.teamId; + if (!workspaceTeamId) return; + const resolved = await resolveWorkspace(workspaceTeamId); + if (!resolved?.botToken) return; - const slack = new WebClient(botToken); + const slack = new WebClient(resolved.botToken); if (!pending) { await slack.chat.update({ @@ -53,7 +74,7 @@ export async function handleSlackInteraction(c: Context) { text: ":x: This action has expired. Please try again.", blocks: [], }); - return c.json({ ok: true }); + return; } if (pending.userId !== userId) { @@ -62,13 +83,13 @@ export async function handleSlackInteraction(c: Context) { user: userId, text: "Only the person who initiated this action can approve or cancel it.", }); - return c.json({ ok: true }); + return; } // Atomic consume — prevents double execution from concurrent requests // (e.g. double-click). If another request already won, return. const consumed = await consume(parsed.pendingId); - if (!consumed) return c.json({ ok: true }); + if (!consumed) return; if (parsed.kind === "cancel") { await slack.chat.update({ @@ -77,7 +98,7 @@ export async function handleSlackInteraction(c: Context) { text: ":no_entry_sign: Cancelled.", blocks: [], }); - return c.json({ ok: true }); + return; } try { @@ -91,7 +112,12 @@ export async function handleSlackInteraction(c: Context) { teamId, }); } catch (err) { - console.error("[slack] action execution error:", err); + logger.error("slack action execution error", { + error: err, + channel: channelId, + teamId, + toolName: consumed.payload.toolName, + }); await slack.chat.update({ channel: channelId, ts: messageTs, @@ -99,8 +125,6 @@ export async function handleSlackInteraction(c: Context) { blocks: [], }); } - - return c.json({ ok: true }); } async function runAndPresent(args: { diff --git a/apps/server/src/routes/slack/registry-runner.ts b/apps/server/src/routes/slack/registry-runner.ts index 8b591b16..20ed0f87 100644 --- a/apps/server/src/routes/slack/registry-runner.ts +++ b/apps/server/src/routes/slack/registry-runner.ts @@ -38,12 +38,16 @@ export function isSlackToolDraft(value: unknown): value is SlackToolDraft { * extraFlag fields (e.g. `notify`) so the model can't be expected to * answer them — the user supplies them via the Block Kit buttons. */ -export function buildSlackTools(ctx: ServiceContext): Record { +export function buildSlackTools( + ctx: ServiceContext, + /** Surface-only tools (e.g. reading the channel the user is viewing). */ + extras?: Record, +): Record { const out: Record = {}; for (const name of Object.keys(agentTools) as AgentToolName[]) { out[name] = buildTool(agentTools[name] as AnyAgentTool, ctx); } - return out; + return { ...out, ...extras }; } export function buildTool(t: AnyAgentTool, ctx: ServiceContext): Tool { diff --git a/apps/server/src/routes/slack/running-turns.ts b/apps/server/src/routes/slack/running-turns.ts new file mode 100644 index 00000000..fac74714 --- /dev/null +++ b/apps/server/src/routes/slack/running-turns.ts @@ -0,0 +1,40 @@ +/** + * Turns currently being processed, so Slack's stop button can cancel one. + * + * Keyed by thread, which is what the user's stop acts on, and what the dedup + * in `handler.ts` already keeps to one turn at a time. + * + * Process-local: with more than one server instance the stop event can land + * where the turn isn't running, and that instance simply finds nothing to + * abort. The handler clears the session status either way, so the user always + * gets out of the loading state. + */ +const turns = new Map(); + +function key(channel: string, threadTs: string): string { + return `${channel}:${threadTs}`; +} + +export function startTurn(channel: string, threadTs: string): AbortController { + const controller = new AbortController(); + turns.set(key(channel, threadTs), controller); + return controller; +} + +export function endTurn( + channel: string, + threadTs: string, + controller: AbortController, +): void { + const id = key(channel, threadTs); + // Only clear our own entry — a newer turn on the same thread keeps its own. + if (turns.get(id) === controller) turns.delete(id); +} + +/** Returns whether a turn was running here to abort. */ +export function abortTurn(channel: string, threadTs: string): boolean { + const controller = turns.get(key(channel, threadTs)); + if (!controller) return false; + controller.abort(); + return true; +} diff --git a/apps/server/src/routes/slack/system-prompt.ts b/apps/server/src/routes/slack/system-prompt.ts index 296f869e..e5d4f55d 100644 --- a/apps/server/src/routes/slack/system-prompt.ts +++ b/apps/server/src/routes/slack/system-prompt.ts @@ -1,6 +1,9 @@ // dependency-free leaf so the prompt test doesn't link the agent's module // graph (ai + services) — bun test flakes on concurrent linking of large graphs. -export function buildSystemPrompt(workspaceName: string): string { +export function buildSystemPrompt( + workspaceName: string, + contextNote?: string, +): string { // Intentional: a per-call ISO timestamp defeats Anthropic/Gateway // prompt caching, but the agent needs minute-level precision to parse // relative times like "next Friday from 2-3 PM". Slack agent volume is @@ -104,5 +107,5 @@ Maintenance scheduling: - Parse natural language dates into ISO 8601 format. Convert relative dates like "next Friday from 2-3 PM" into proper ISO 8601 timestamps. - If the user doesn't specify a timezone, default to UTC and mention that in your response. - The "from" time must be before the "to" time. -- Write a professional maintenance message describing what will happen during the window.`; +- Write a professional maintenance message describing what will happen during the window.${contextNote ?? ""}`; } diff --git a/apps/server/src/routes/slack/thread-title.test.ts b/apps/server/src/routes/slack/thread-title.test.ts new file mode 100644 index 00000000..22fd67e7 --- /dev/null +++ b/apps/server/src/routes/slack/thread-title.test.ts @@ -0,0 +1,106 @@ +import { beforeEach, describe, expect, test } from "@openstatus/test-utils"; + +import { + buildThreadTitle, + isThreadTitled, + markThreadTitled, + truncateTitle, +} from "./thread-title"; + +const redisStore = (globalThis as Record) + .__testRedisStore as Map; + +describe("buildThreadTitle", () => { + test("prefers the drafted report title", () => { + expect( + buildThreadTitle({ + draft: { + toolName: "create_status_report", + input: { title: "Elevated API error rates", status: "investigating" }, + }, + userText: "<@UBOT> we're seeing errors, open a report", + }), + ).toBe("Elevated API error rates"); + }); + + test("marks maintenance so it doesn't read as an outage", () => { + expect( + buildThreadTitle({ + draft: { + toolName: "create_maintenance", + input: { title: "Database upgrade" }, + }, + }), + ).toBe("Maintenance: Database upgrade"); + }); + + test("falls back to the report an update acts on", () => { + // `add_status_report_update` carries an id, never a title. + expect( + buildThreadTitle({ + draft: { + toolName: "add_status_report_update", + input: { statusReportId: 42, status: "identified" }, + }, + reportTitle: "Checkout latency in fra", + userText: "<@UBOT> we found the cause", + }), + ).toBe("Checkout latency in fra"); + }); + + test("uses what the user asked when the turn writes nothing", () => { + expect( + buildThreadTitle({ userText: "<@UBOT> what's broken right now? " }), + ).toBe("what's broken right now?"); + }); + + test("gives up rather than inventing a name", () => { + expect(buildThreadTitle({})).toBeUndefined(); + expect(buildThreadTitle({ userText: " " })).toBeUndefined(); + expect(buildThreadTitle({ userText: "<@UBOT>" })).toBeUndefined(); + // A draft without a usable title falls through, not onto `[object Object]`. + expect( + buildThreadTitle({ + draft: { toolName: "resolve_status_report", input: { title: " " } }, + }), + ).toBeUndefined(); + }); +}); + +describe("truncateTitle", () => { + test("leaves a short title alone and collapses whitespace", () => { + expect(truncateTitle(" Elevated API errors\n")).toBe( + "Elevated API errors", + ); + }); + + test("cuts on a word boundary", () => { + const long = + "Elevated error rates affecting checkout and billing across every region"; + const result = truncateTitle(long); + expect(result.length).toBeLessThanOrEqual(61); + expect(result.endsWith("…")).toBe(true); + expect(result).toBe( + "Elevated error rates affecting checkout and billing across…", + ); + }); + + test("hard-cuts a single word that runs past the limit", () => { + const result = truncateTitle("x".repeat(90)); + expect(result).toBe(`${"x".repeat(60)}…`); + }); +}); + +describe("the titled marker", () => { + beforeEach(() => redisStore.clear()); + + test("is absent until set, then reported for that thread only", async () => { + expect(await isThreadTitled("D1", "1.1")).toBe(false); + + await markThreadTitled("D1", "1.1"); + + expect(await isThreadTitled("D1", "1.1")).toBe(true); + expect(await isThreadTitled("D1", "2.2")).toBe(false); + expect(await isThreadTitled("D2", "1.1")).toBe(false); + }); +}); diff --git a/apps/server/src/routes/slack/thread-title.ts b/apps/server/src/routes/slack/thread-title.ts new file mode 100644 index 00000000..72d38c66 --- /dev/null +++ b/apps/server/src/routes/slack/thread-title.ts @@ -0,0 +1,138 @@ +import { getLogger } from "@logtape/logtape"; +import type { WebClient } from "@slack/web-api"; + +import { redis } from "@/libs/clients"; + +const logger = getLogger("api-server"); + +// Slack accepts 200 characters, but the Messages tab shows far fewer before +// it elides — a title that only reads in full on hover isn't doing its job. +const MAX_TITLE = 60; + +/** + * Presence means the thread's title is settled: either we set it on the first + * turn that produced a subject, or a person renamed it by hand. Both mean the + * same thing here — leave it alone. A thread is one incident's whole life, so + * the name should be its subject, not whatever the latest turn happened to do. + * + * The TTL outlives any incident by a wide margin; if it ever does lapse, the + * cost is one stale thread getting re-titled on a much later turn. + */ +const TITLE_PREFIX = "slack:title:"; +const TITLE_TTL_SECONDS = 30 * 24 * 60 * 60; + +function titleKey(channel: string, threadTs: string): string { + return `${TITLE_PREFIX}${channel}:${threadTs}`; +} + +export async function isThreadTitled( + channel: string, + threadTs: string, +): Promise { + return (await redis.get(titleKey(channel, threadTs))) !== null; +} + +export async function markThreadTitled( + channel: string, + threadTs: string, +): Promise { + await redis.set(titleKey(channel, threadTs), "1", { + ex: TITLE_TTL_SECONDS, + }); +} + +/** Cuts on a word boundary, falling back to a hard cut for one long word. */ +export function truncateTitle(text: string): string { + const clean = text.replace(/\s+/g, " ").trim(); + if (clean.length <= MAX_TITLE) return clean; + const cut = clean.slice(0, MAX_TITLE); + const lastSpace = cut.lastIndexOf(" "); + const kept = lastSpace > MAX_TITLE / 2 ? cut.slice(0, lastSpace) : cut; + return `${kept.trimEnd()}…`; +} + +export interface TitleSources { + /** The write this turn proposed, if any. */ + draft?: { toolName: string; input: unknown }; + /** Title of the report the draft acts on, for updates and resolutions. */ + reportTitle?: string; + /** What the user said, mention included — it gets stripped here. */ + userText?: string; +} + +/** + * The thread's subject, best source first: + * + * 1. a drafted report or maintenance title — already written for people to + * read, since it is what lands on the status page; + * 2. the title of the report an update or resolution acts on; + * 3. what the user asked, for turns that produce nothing else. + * + * Undefined when none of those yield anything: Slack's own default name beats + * a title we made up. + */ +export function buildThreadTitle({ + draft, + reportTitle, + userText, +}: TitleSources): string | undefined { + const drafted = draft && draftTitle(draft); + if (drafted) return truncateTitle(drafted); + if (reportTitle?.trim()) return truncateTitle(reportTitle); + const asked = strippedUserText(userText); + return asked ? truncateTitle(asked) : undefined; +} + +function draftTitle(draft: { + toolName: string; + input: unknown; +}): string | undefined { + if (typeof draft.input !== "object" || draft.input === null) return undefined; + const title = (draft.input as { title?: unknown }).title; + if (typeof title !== "string" || !title.trim()) return undefined; + // Planned work reads as an outage in a timeline unless it says otherwise. + return draft.toolName === "create_maintenance" + ? `Maintenance: ${title}` + : title; +} + +function strippedUserText(text: string | undefined): string | undefined { + if (!text) return undefined; + const stripped = text + .replace(/<@[A-Z0-9]+>/g, " ") + .replace(/\s+/g, " ") + .trim(); + return stripped || undefined; +} + +/** + * Names the thread, once. Two turns racing here would both rename to the same + * subject, so the check-then-set is left unguarded. + */ +export async function renameThread(args: { + slack: WebClient; + channel: string; + threadTs: string; + title: string; + teamId: string; +}): Promise { + const { slack, channel, threadTs, title, teamId } = args; + try { + await slack.agents.sessions.rename({ + channel_id: channel, + thread_ts: threadTs, + title, + }); + } catch (err) { + // Workspaces without agent sessions throw here; the thread simply keeps + // the name Slack gives it. + logger.info("slack could not rename the thread", { + error: err, + channel, + teamId, + }); + return; + } + await markThreadTitled(channel, threadTs); + logger.info("slack thread titled", { channel, threadTs, teamId, title }); +}