diff --git a/apps/slack-agent/README.md b/apps/slack-agent/README.md index eec7e0375..8f8a62e70 100644 --- a/apps/slack-agent/README.md +++ b/apps/slack-agent/README.md @@ -153,8 +153,9 @@ settings: bot_events: - app_mention - message.im - # Thread follow-ups without re-mentioning the bot: replies in threads the - # bot is engaged in are promoted to app_mention by the webhookVerifier + # Thread follow-ups without re-mentioning the bot: candidate thread + # replies are promoted to app_mention by the webhookVerifier, and + # confirmed (or dropped) after the 200 by the mention handler # (agent/lib/thread-follow-up.ts). Only channels the bot is a member of # deliver these events. - message.channels @@ -288,11 +289,17 @@ Both should show a green **Verified ✓** next to the field once saved. Event Su `app_mention`, `message.im`, `message.channels`, and `message.groups` listed under _Subscribe to bot events_ — the manifest from step 1 sets these, so they should already be there. The two channel message events power thread follow-ups: once the bot has been mentioned (or replied) in a thread, -further replies in that thread reach it without a new `@mention` — but only while the engagement is -**recent**: within 30 minutes and within the last 15 messages of the thread (see -`agent/lib/thread-follow-up.ts`). Past either bound, replies pass through untouched and the user -@-mentions the bot again. Unbounded, one mention would turn every later reply by anyone into a full -agent turn, forever. +further replies in that thread reach it without a new `@mention` — but only while the thread is +still **alive** (nobody has touched it for less than 24 hours) and the bot is still within its +**last 15 messages**. Past either bound the reply is dropped, everyone who spoke in the thread gets +a DM saying so, and one `@mention` brings the bot back (`agent/lib/thread-follow-up.ts`, +`agent/lib/disengage-notice.ts`). Unbounded, one mention would turn every later reply by anyone into +a full agent turn, forever. + +The decision deliberately does **not** happen in the webhook verifier — that is the only awaited work +before eve's 200, so anything it fetches is spent out of Slack's ~3s delivery budget. Promotion there +is parse-only and optimistic; the mention handler confirms it afterwards, against the thread it loads +for turn context anyway. Changing a request URL does **not** require reinstalling the app; only changing _scopes_ does. (If you did edit scopes, the sidebar shows a yellow reinstall banner — follow it, and note that diff --git a/apps/slack-agent/agent/channels/slack.ts b/apps/slack-agent/agent/channels/slack.ts index 2c5f89f96..190683c4f 100644 --- a/apps/slack-agent/agent/channels/slack.ts +++ b/apps/slack-agent/agent/channels/slack.ts @@ -1,12 +1,19 @@ import { defaultSlackAuth, slackChannel } from "eve/channels/slack" import type { SlackContext, SlackMentionResult, SlackMessage, SlackWebhookVerifier } from "eve/channels/slack" -import { acknowledgeIncomingMessage } from "#lib/ack-reaction.js" +import { acknowledgeIncomingMessage, acknowledgeMessage } from "#lib/ack-reaction.js" import { describeActions, truncateTypingStatus } from "#lib/action-status.js" import { botUserIdForTeam, rememberBotUserId } from "#lib/bot-identity.js" import { loadChannelContext } from "#lib/channel-context.js" +import { notifyThreadDisengagement } from "#lib/disengage-notice.js" import { resolveBotToken, verifySlackV0Signature, type SlackTokenContext } from "#lib/maple.js" -import { loadThreadContext } from "#lib/thread-context.js" -import { promoteThreadFollowUp, recordThreadEngagement } from "#lib/thread-follow-up.js" +import { emitAgentLog } from "#lib/telemetry-log.js" +import { formatThreadContext, loadThreadMessages } from "#lib/thread-context.js" +import { + confirmThreadFollowUp, + pendingFollowUp, + promoteThreadFollowUp, + recordThreadEngagement, +} from "#lib/thread-follow-up.js" import { formatTurnTime } from "#lib/turn-time.js" import { forwardUninstallEvent } from "#lib/uninstall-detection.js" @@ -48,10 +55,9 @@ const webhookVerifier: SlackWebhookVerifier = async (request, body) => { rememberBotUserId(body) // Learn engagement from events that already prove it — the bot's own posts - // echoing back, and @mentions of it — so the promotion below can answer from - // cache instead of spending Slack's webhook budget on `conversations.replies` - // in a thread the bot is demonstrably active in (#lib/thread-follow-up.js). - // Synchronous and network-free; must run before the promotion. + // echoing back, and @mentions of it — so the confirmation step downstream can + // answer from memory in a thread the bot is demonstrably active in + // (#lib/thread-follow-up.js). Synchronous and network-free. recordThreadEngagement(body) // app_uninstalled / tokens_revoked: eve only dispatches app_mention + DM @@ -61,34 +67,39 @@ const webhookVerifier: SlackWebhookVerifier = async (request, body) => { void forwardUninstallEvent(body) // Instant "received" ack: react with :eyes: on any message eve will - // dispatch as an agent turn (mentions + DMs), before the turn is even + // dispatch as an agent turn (real mentions + DMs), before the turn is even // scheduled. Fired without awaiting — never delays the webhook ack. // Slack redelivery retries skip it (`already_reacted` is also tolerated // downstream, this just avoids the pointless call). + // + // Promoted follow-ups are deliberately NOT ack'd here. Their promotion is + // optimistic and `dispatchWithConversationContext` may still drop them, and + // acking a message we then never answer is worse than not acking at all: the + // :eyes: is a promise. They get their ack there, once the promotion is + // confirmed. const isSlackRetry = request.headers.get("x-slack-retry-num") !== null if (!isSlackRetry) void acknowledgeIncomingMessage(body) // eve parses whatever body we return, which is also our hook for thread // follow-ups: eve only dispatches app_mention + DM events, so an un-mentioned - // reply in a thread the bot is engaged in gets its `event.type` promoted to - // "app_mention" here (see #lib/thread-follow-up.js). Everything else passes - // through verified-but-unchanged. + // reply in a thread that could be a follow-up gets its `event.type` promoted + // to "app_mention" here (see #lib/thread-follow-up.js). Everything else + // passes through verified-but-unchanged. + // + // This is the LAST awaited work before eve returns 200, i.e. the only thing + // still spending Slack's ~3s delivery budget — which is why the promotion is + // parse-only and synchronous, and why the engagement decision itself now + // happens after the 200. It cannot throw, but a throw here would fail every + // inbound event, so the guard stays. try { - const promoted = await promoteThreadFollowUp(body) - if (promoted !== null) { - // A promoted follow-up is agent work too, but its raw body (a plain - // channel `message`) doesn't qualify above — ack it now that we know - // the bot is engaged. - if (!isSlackRetry) void acknowledgeIncomingMessage(promoted) - return promoted - } + return promoteThreadFollowUp(body, { isSlackRetry }) ?? body } catch (error) { console.warn( "[slack-webhook] Thread follow-up promotion failed; passing the event through unchanged.", error, ) + return body } - return body } /** @@ -116,25 +127,104 @@ const webhookVerifier: SlackWebhookVerifier = async (request, body) => { * the alert's window instead of the current one. * * Runs after eve has already returned 200 to Slack (`waitUntil`), so the Slack - * fetches are off the webhook's delivery budget. It must not throw: eve drops - * the whole mention when this handler does, so both loads degrade to no context - * instead. + * fetches are off the webhook's delivery budget. That is also why the thread + * follow-up decision lives here rather than in the verifier: this handler + * already loads the thread, and here it can take as long as it needs. It must + * not otherwise throw: eve drops the whole mention when this handler does, so + * both loads degrade to no context instead — the one deliberate throw is the + * disengagement drop below, which is exactly what that escape hatch is for. */ async function dispatchWithConversationContext( ctx: SlackContext, message: SlackMessage, ): Promise { - await ctx.thread.startTyping("Thinking...") - const [threadContext, channelContext] = await Promise.all([ - loadThreadContext(ctx.thread, message, { botUserId: botUserIdForTeam(message.teamId) }), + // Non-null only for a follow-up the verifier promoted optimistically — the + // one kind of dispatch that still has to earn its turn. Registry lookup, not + // a text sniff: eve re-renders inbound mrkdwn, so `message.text` is no longer + // proof of what Slack actually sent (#lib/thread-follow-up.js). + const pending = pendingFollowUp({ + teamId: message.teamId, + channelId: message.channelId, + messageTs: message.ts, + }) + + // A real mention is answered for certain, so it gets the typing indicator + // immediately. A follow-up waits until it is confirmed: "Thinking..." on a + // message the bot then silently drops reads worse than saying nothing. + if (pending === null) await ctx.thread.startTyping("Thinking...") + + const botUserId = botUserIdForTeam(message.teamId) + const [threadMessages, channelContext] = await Promise.all([ + loadThreadMessages(ctx.thread, message), loadChannelContext(message), ]) + + if (pending !== null) { + const decision = confirmThreadFollowUp(pending, threadMessages) + if (!decision.engaged) { + // The drop is silent by construction — no turn, no reply, no reaction — + // so this log is the only trace it leaves. hooks/outcome-log.ts cannot + // cover it: there is no turn for it to report on. Ids only, never text. + // + // Severity splits on the same distinction the DM does: declining a + // thread the bot was never in is the ordinary outcome of promoting + // optimistically (most channel replies are not follow-ups), while + // leaving a conversation the bot WAS in is a user who did not get an + // answer, and worth finding later. + emitAgentLog(decision.workedInThread ? "warn" : "info", "follow_up_disengaged", { + "maple.agent.event": "follow_up_disengaged", + "maple.agent.disengage_reason": decision.reason, + "maple.agent.worked_in_thread": decision.workedInThread, + "maple.slack.team_id": pending.teamId, + "maple.slack.channel_id": pending.channelId, + "maple.slack.thread_ts": pending.threadTs, + "maple.slack.message_ts": pending.messageTs, + }) + // Only for a thread the bot actually worked in: leaving those people + // wondering why it stopped answering is the failure this fixes. A thread + // it was never in is ordinary channel chatter and must stay silent. + if (decision.workedInThread && threadMessages !== null) { + void notifyThreadDisengagement({ + teamId: pending.teamId, + channelId: pending.channelId, + threadTs: pending.threadTs, + messageTs: pending.messageTs, + reason: decision.reason, + messages: threadMessages, + replierUserId: message.author?.userId, + botUserId, + }) + } + // eve's `dispatchInboundMessage` catches a throwing handler and abandons + // the turn before the model sees anything — the only way to un-dispatch + // an event it has already accepted. + throw new Error( + `Thread follow-up not dispatched (${decision.reason}): the bot is no longer engaged in ${pending.channelId}:${pending.threadTs}.`, + ) + } + // Confirmed, so the :eyes: is now a promise we keep. + if (pending.ackable) { + void acknowledgeMessage({ + teamId: pending.teamId, + channelId: pending.channelId, + messageTs: pending.messageTs, + threadTs: pending.threadTs, + }) + } + await ctx.thread.startTyping("Thinking...") + } + + const threadContext = formatThreadContext(threadMessages ?? [], { botUserId }) // Channel first: it is the background the thread happens in. The clock last, // so it sits closest to the message being answered — and so that in a thread // with several turns' worth of these, the newest is the one nearest the ask. return { auth: defaultSlackAuth(message, ctx), - context: [...channelContext, ...threadContext, formatTurnTime(message)], + context: [ + ...channelContext, + ...(threadContext === undefined ? [] : [threadContext]), + formatTurnTime(message), + ], } } diff --git a/apps/slack-agent/agent/lib/ack-reaction.ts b/apps/slack-agent/agent/lib/ack-reaction.ts index 4a9ec326d..de374ce1c 100644 --- a/apps/slack-agent/agent/lib/ack-reaction.ts +++ b/apps/slack-agent/agent/lib/ack-reaction.ts @@ -13,12 +13,16 @@ import { ACK_REACTION_NAME, addReactionViaSlack, registerAckedTriggeringMessage * throws — a failed reaction is cosmetic, the turn still runs. * * It reacts exactly to the bodies eve will dispatch as agent work: - * - `app_mention` events (including thread follow-ups promoted to - * `app_mention` by `promoteThreadFollowUp` — call this on the PROMOTED - * body; the unpromoted twin does not qualify, so no double reaction); + * - real `app_mention` events; * - user-authored DM `message` events (`channel_type: "im"`, no `bot_id`, * no subtype except `file_share` — mirrors eve's own DM dispatch filter). * + * Thread follow-ups promoted to `app_mention` by `promoteThreadFollowUp` are + * NOT ack'd from the webhook. That promotion is optimistic and the handler may + * still drop the message, and a `:eyes:` on a message the bot never answers is + * a promise broken in public — so `#channels/slack.js` acks them through + * `acknowledgeMessage` once the engagement check has confirmed the turn. + * * Requires the Slack app's `reactions:write` scope. * * Each qualifying message is also registered in lib/reaction.ts's @@ -107,10 +111,23 @@ export function parseAckReactionTarget(rawBody: string): AckReactionTarget | nul export async function acknowledgeIncomingMessage( rawBody: string, deps: AckReactionDeps = defaultDeps, +): Promise { + const target = parseAckReactionTarget(rawBody) + if (!target) return + await acknowledgeMessage(target, deps) +} + +/** + * Reacts with `:eyes:` on an already-identified message. The entry point for + * callers that no longer hold a raw webhook body — the thread-follow-up dispatch + * path, which only learns the turn is really happening after eve has parsed the + * event away. Never throws. + */ +export async function acknowledgeMessage( + target: AckReactionTarget, + deps: AckReactionDeps = defaultDeps, ): Promise { try { - const target = parseAckReactionTarget(rawBody) - if (!target) return // Registered before the reaction call: the `add_reaction` tool needs the // triggering message's ts even when this ack itself fails (its remove of // a never-added `:eyes:` is tolerated downstream). diff --git a/apps/slack-agent/agent/lib/disengage-notice.test.ts b/apps/slack-agent/agent/lib/disengage-notice.test.ts new file mode 100644 index 000000000..ca802b160 --- /dev/null +++ b/apps/slack-agent/agent/lib/disengage-notice.test.ts @@ -0,0 +1,343 @@ +import { beforeEach, describe, expect, test } from "bun:test" +import type { SlackThreadMessage } from "eve/channels/slack" +import { + disengagementNoticeText, + getPermalinkFromSlack, + humanThreadParticipants, + noticeRecipients, + notifyThreadDisengagement, + postDirectMessageViaSlack, + resetDisengageNoticeStateForTests, + type DisengageNoticeDeps, + type ThreadDisengagement, +} from "./disengage-notice.js" +import { installFetchStub } from "./fetch-stub.js" + +const BOT_USER_ID = "U0BOT" +const PERMALINK = "https://acme.slack.com/archives/C123/p1700000002000200" + +const threadMessage = (overrides: Partial = {}): SlackThreadMessage => ({ + text: "", + markdown: "", + user: undefined, + botId: undefined, + ts: "1700000000.000100", + threadTs: "1700000000.000100", + isMe: false, + raw: {}, + ...overrides, +}) + +const humanMessage = (user: string, ts: string): SlackThreadMessage => + threadMessage({ text: "…", markdown: "…", user, ts }) + +const disengagement = (overrides: Partial = {}): ThreadDisengagement => ({ + teamId: "T123", + channelId: "C123", + threadTs: "1700000000.000100", + messageTs: "1700000002.000200", + reason: "thread-dormant", + messages: [humanMessage("U456", "1700000000.000100"), humanMessage("U789", "1700000001.000100")], + replierUserId: "U456", + botUserId: BOT_USER_ID, + ...overrides, +}) + +function makeDeps(overrides: Partial = {}): { + deps: DisengageNoticeDeps + dms: { userId: string; text: string }[] +} { + const dms: { userId: string; text: string }[] = [] + return { + dms, + deps: { + resolveBotToken: async () => "xoxb-test", + getPermalink: async () => PERMALINK, + postDirectMessage: async ({ userId, text }) => { + dms.push({ userId, text }) + }, + ...overrides, + }, + } +} + +beforeEach(() => { + resetDisengageNoticeStateForTests() +}) + +// ── recipients ────────────────────────────────────────────────────────────── + +describe("humanThreadParticipants", () => { + test("collects every distinct human in the thread, oldest first", () => { + expect( + humanThreadParticipants( + disengagement({ + messages: [ + humanMessage("U456", "1700000000.000100"), + humanMessage("U789", "1700000001.000100"), + humanMessage("U456", "1700000002.000100"), + ], + }), + ), + ).toEqual(["U456", "U789"]) + }) + + test("includes the author of the dropped reply, whose message is not in the thread", () => { + // `loadThreadContextMessages` excludes the triggering message, so the one + // person guaranteed to want this DM would otherwise be the one to miss it. + expect( + humanThreadParticipants( + disengagement({ + messages: [humanMessage("U789", "1700000000.000100")], + replierUserId: "U456", + }), + ), + ).toEqual(["U789", "U456"]) + }) + + test("excludes the bot itself and every other app", () => { + const alertCard = threadMessage({ user: BOT_USER_ID, botId: "B0BOT", ts: "1700000000.000100" }) + const github = threadMessage({ user: "U0GH", botId: "B0GH", ts: "1700000001.000100" }) + expect( + humanThreadParticipants( + disengagement({ + messages: [alertCard, github, humanMessage("U456", "1700000002.000100")], + replierUserId: "U456", + }), + ), + ).toEqual(["U456"]) + }) + + test("skips messages Slack attributed to nobody", () => { + expect( + humanThreadParticipants( + disengagement({ + messages: [threadMessage({ ts: "1700000000.000100" })], + replierUserId: undefined, + }), + ), + ).toEqual([]) + }) +}) + +// ── the notice ────────────────────────────────────────────────────────────── + +// The two reasons describe opposite situations, so they cannot share one +// fan-out rule: `thread-dormant` is a quiet thread nobody is mid-discussion in, +// while `engagement-buried` is a *lively* one that ran past the bot — DMing +// everyone there reaches people who moved on, about someone else's message. + +describe("noticeRecipients", () => { + test("a dormant thread notifies everyone who took part", () => { + expect(noticeRecipients(disengagement({ reason: "thread-dormant" }))).toEqual(["U456", "U789"]) + }) + + test("a buried engagement notifies only the author of the unanswered reply", () => { + expect( + noticeRecipients(disengagement({ reason: "engagement-buried", replierUserId: "U789" })), + ).toEqual(["U789"]) + }) + + test("the replier is notified even when this is their first message in the thread", () => { + // Their message is the one that triggered this, so it is not in `messages`. + expect( + noticeRecipients( + disengagement({ + reason: "engagement-buried", + replierUserId: "U999", + messages: [humanMessage("U456", "1700000000.000100")], + }), + ), + ).toEqual(["U999"]) + }) + + test("an unknown replier falls back to everyone rather than telling nobody", () => { + expect( + noticeRecipients(disengagement({ reason: "engagement-buried", replierUserId: undefined })), + ).toEqual(["U456", "U789"]) + }) + + test("a replier that resolves to the bot itself falls back to everyone", () => { + expect( + noticeRecipients( + disengagement({ reason: "engagement-buried", replierUserId: BOT_USER_ID }), + ), + ).toEqual(["U456", "U789"]) + }) +}) + +describe("notifyThreadDisengagement", () => { + test("a buried engagement DMs only the replier, not the whole thread", async () => { + const { deps, dms } = makeDeps() + await notifyThreadDisengagement( + disengagement({ reason: "engagement-buried", replierUserId: "U789" }), + deps, + ) + expect(dms.map((d) => d.userId)).toEqual(["U789"]) + }) + + test("DMs every human in the thread, with a permalink and how to resume", async () => { + const { deps, dms } = makeDeps() + await notifyThreadDisengagement(disengagement(), deps) + + expect(dms.map((dm) => dm.userId)).toEqual(["U456", "U789"]) + for (const dm of dms) { + expect(dm.text).toContain(PERMALINK) + expect(dm.text).toContain("@-mention me") + expect(dm.text).toContain("quiet for over a day") + } + }) + + test("the permalink points at the reply that went unanswered", async () => { + let asked: { channelId: string; messageTs: string } | undefined + const { deps } = makeDeps({ + getPermalink: async (options) => { + asked = { channelId: options.channelId, messageTs: options.messageTs } + return PERMALINK + }, + }) + await notifyThreadDisengagement(disengagement(), deps) + expect(asked).toEqual({ channelId: "C123", messageTs: "1700000002.000200" }) + }) + + test("a buried engagement is explained differently from a dormant thread", async () => { + const { deps, dms } = makeDeps() + await notifyThreadDisengagement(disengagement({ reason: "engagement-buried" }), deps) + expect(dms[0]?.text).toContain("moved well past") + }) + + test("notifies once per thread, however many replies follow", async () => { + // Otherwise every later reply in a thread the bot has left DMs everyone + // again — a helpful nudge turned into the noise that gets an app removed. + const { deps, dms } = makeDeps() + await notifyThreadDisengagement(disengagement(), deps) + await notifyThreadDisengagement(disengagement({ messageTs: "1700000003.000300" }), deps) + expect(dms.length).toBe(2) + }) + + test("other threads are unaffected", async () => { + const { deps, dms } = makeDeps() + await notifyThreadDisengagement(disengagement(), deps) + await notifyThreadDisengagement(disengagement({ threadTs: "1700000009.000100" }), deps) + expect(dms.length).toBe(4) + }) + + test("the guard is per workspace: Slack channel ids are only unique per team", async () => { + const { deps, dms } = makeDeps() + await notifyThreadDisengagement(disengagement(), deps) + await notifyThreadDisengagement(disengagement({ teamId: "T999" }), deps) + expect(dms.length).toBe(4) + }) + + test("sends nothing when there is nobody to tell", async () => { + const { deps, dms } = makeDeps() + await notifyThreadDisengagement(disengagement({ messages: [], replierUserId: undefined }), deps) + expect(dms.length).toBe(0) + }) + + test("still sends when Slack cannot produce a permalink", async () => { + const { deps, dms } = makeDeps({ getPermalink: async () => null }) + await notifyThreadDisengagement(disengagement(), deps) + expect(dms.length).toBe(2) + expect(dms[0]?.text).toContain("a Slack thread") + }) + + test("a permalink failure does not cost the notice", async () => { + const { deps, dms } = makeDeps({ + getPermalink: async () => { + throw new Error("slack down") + }, + }) + await expect(notifyThreadDisengagement(disengagement(), deps)).resolves.toBeUndefined() + expect(dms.length).toBe(2) + }) + + test("one closed DM does not cost the others theirs", async () => { + const delivered: string[] = [] + const { deps } = makeDeps({ + postDirectMessage: async ({ userId }) => { + if (userId === "U456") throw new Error("cannot_dm_bot") + delivered.push(userId) + }, + }) + await expect(notifyThreadDisengagement(disengagement(), deps)).resolves.toBeUndefined() + expect(delivered).toEqual(["U789"]) + }) + + test("never throws: it is fired without awaiting and nothing is waiting on it", async () => { + const { deps } = makeDeps({ + resolveBotToken: async () => { + throw new Error("team not linked") + }, + }) + await expect(notifyThreadDisengagement(disengagement(), deps)).resolves.toBeUndefined() + }) +}) + +describe("disengagementNoticeText", () => { + test("reads as a sentence without the link, for the notification preview", () => { + expect(disengagementNoticeText("thread-dormant", null)).toContain( + "There's a new reply in a Slack thread I'd been working in", + ) + }) +}) + +// ── Slack request shapes ──────────────────────────────────────────────────── + +describe("Slack calls", () => { + test("chat.getPermalink asks for the channel and the message", async () => { + const stub = installFetchStub(() => Response.json({ ok: true, permalink: PERMALINK })) + try { + const permalink = await getPermalinkFromSlack({ + botToken: "xoxb-test", + channelId: "C123", + messageTs: "1700000002.000200", + }) + expect(permalink).toBe(PERMALINK) + const url = new URL(String(stub.calls[0]?.url)) + expect(url.pathname).toBe("/api/chat.getPermalink") + expect(url.searchParams.get("channel")).toBe("C123") + expect(url.searchParams.get("message_ts")).toBe("1700000002.000200") + } finally { + stub.restore() + } + }) + + test("chat.getPermalink returns null rather than throwing when Slack refuses", async () => { + const stub = installFetchStub(() => Response.json({ ok: false, error: "message_not_found" })) + try { + expect( + await getPermalinkFromSlack({ + botToken: "xoxb-test", + channelId: "C123", + messageTs: "1700000002.000200", + }), + ).toBeNull() + } finally { + stub.restore() + } + }) + + test("chat.postMessage addresses the user id directly — Slack opens the DM", async () => { + const stub = installFetchStub(() => Response.json({ ok: true })) + try { + await postDirectMessageViaSlack({ botToken: "xoxb-test", userId: "U456", text: "hi" }) + expect(stub.calls[0]?.url).toBe("https://slack.com/api/chat.postMessage") + expect(JSON.parse(String(stub.calls[0]?.body))).toEqual({ channel: "U456", text: "hi" }) + expect(stub.calls[0]?.headers.authorization).toBe("Bearer xoxb-test") + } finally { + stub.restore() + } + }) + + test("chat.postMessage surfaces a Slack-level failure to the per-recipient catch", async () => { + const stub = installFetchStub(() => Response.json({ ok: false, error: "channel_not_found" })) + try { + await expect( + postDirectMessageViaSlack({ botToken: "xoxb-test", userId: "U456", text: "hi" }), + ).rejects.toThrow("channel_not_found") + } finally { + stub.restore() + } + }) +}) diff --git a/apps/slack-agent/agent/lib/disengage-notice.ts b/apps/slack-agent/agent/lib/disengage-notice.ts new file mode 100644 index 000000000..082611b32 --- /dev/null +++ b/apps/slack-agent/agent/lib/disengage-notice.ts @@ -0,0 +1,306 @@ +import type { SlackThreadMessage } from "eve/channels/slack" +import { resolveBotToken } from "./maple.js" +import type { FollowUpDisengagedReason } from "./thread-follow-up.js" +import { createTtlCache } from "./ttl-cache.js" + +/** + * Telling people the bot has left a thread. + * + * Thread follow-ups (`#lib/thread-follow-up.js`) let users keep talking to the + * bot without re-@-mentioning it — until one of the disengagement bounds trips, + * at which point the reply is dropped. That drop is *correct* and it is also + * completely invisible: no reply, no `:eyes:`, no error. The user sees a bot + * that answered them ten messages ago and is now ignoring them, and has no way + * to know that one @mention would bring it straight back. + * + * So a disengagement that ends a conversation the bot was actually part of gets + * a DM. The distinction matters: a thread the bot has never been in is ordinary + * channel chatter, and DMing about it would be the bot introducing itself to + * strangers, uninvited. `confirmThreadFollowUp` reports which case it is + * (`workedInThread`), which it can only do now that the engagement check reads + * the whole thread rather than a window that began after the bot's last post. + * + * Everyone who spoke in the thread is notified, not only the person whose reply + * was dropped: in a thread the bot was working in, the whole group loses its + * answer, and whoever asks next should not have to rediscover the same silence. + * + * Fired without awaiting and never throws — same contract as + * `forwardUninstallEvent` / `acknowledgeIncomingMessage`. A notice that fails to + * send must not cost anything else; the thread is already not getting a turn. + * + * Scopes: `im:write` (open the DM) and `chat:write` (post it), both already + * granted — no reinstall. + */ + +const SLACK_API = "https://slack.com/api" + +/** + * A stalled Slack API must not keep a fire-and-forget notice (and the bot token + * it holds) alive indefinitely. + */ +const SLACK_API_TIMEOUT_MS = 10_000 + +/** + * One notice per thread per disengagement episode. + * + * Without this, every subsequent reply in a thread the bot has left would DM + * everyone again — turning a helpful nudge into exactly the kind of noise that + * gets an app uninstalled. + * + * The TTL matches the dormancy bound in `#lib/thread-follow-up.js` on purpose: + * a thread can only re-enter dormancy after another full day of silence, so one + * day of suppression is precisely one episode. A thread that disengaged the + * other way (the conversation ran past the bot) is even better served — it is + * busy, so it would otherwise re-notify the most. + */ +const NOTICE_TTL_MS = 24 * 60 * 60_000 +const NOTICE_CACHE_MAX_ENTRIES = 500 +const NOTICE_CACHE_SWEEP_INTERVAL_MS = 60_000 + +interface NoticeEntry { + readonly expiresAt: number +} + +const noticedThreads = createTtlCache({ + maxEntries: NOTICE_CACHE_MAX_ENTRIES, + sweepIntervalMs: NOTICE_CACHE_SWEEP_INTERVAL_MS, +}) + +/** Test-only: forgets which threads have been notified. */ +export function resetDisengageNoticeStateForTests(): void { + noticedThreads.clear() +} + +/** + * Process-global cache across every tenant, and Slack channel ids are only + * unique per workspace — the team has to be in the key. + */ +function noticeKey(teamId: string | undefined, channelId: string, threadTs: string): string { + return `${teamId ?? "-"}:${channelId}:${threadTs}` +} + +export interface ThreadDisengagement { + readonly teamId: string | undefined + readonly channelId: string + readonly threadTs: string + /** Slack ts of the reply that went unanswered — where the permalink points. */ + readonly messageTs: string + readonly reason: FollowUpDisengagedReason + /** The thread as the handler loaded it; the notice's recipient list. */ + readonly messages: readonly SlackThreadMessage[] + /** Author of the dropped reply. Always notified, even if this is their first message. */ + readonly replierUserId: string | undefined + /** This workspace's bot user id (`#lib/bot-identity.js`), so it can exclude itself. */ + readonly botUserId: string | undefined +} + +/** Injectable dependencies so tests never touch the network. */ +export interface DisengageNoticeDeps { + resolveBotToken(context: { teamId?: string }): Promise + getPermalink(options: { + readonly botToken: string + readonly channelId: string + readonly messageTs: string + }): Promise + postDirectMessage(options: { + readonly botToken: string + /** Slack user id: `chat.postMessage` opens the DM itself when `channel` is one. */ + readonly userId: string + readonly text: string + }): Promise +} + +export const defaultDisengageNoticeDeps: DisengageNoticeDeps = { + resolveBotToken, + getPermalink: getPermalinkFromSlack, + postDirectMessage: postDirectMessageViaSlack, +} + +/** + * The distinct humans who spoke in the thread, oldest first, with the author of + * the dropped reply guaranteed to be among them (their message is not part of + * the loaded thread — it is the one that triggered this). + * + * Bots are excluded, this one included. `botId` is the right filter rather than + * a user-id comparison alone: Maple's own alert cards are posted through this + * bot user and are nobody's message, and a GitHub or CI app's posts are nobody + * to DM either. + * + * Exported for the tests; `notifyThreadDisengagement` calls it itself. + */ +export function humanThreadParticipants(disengagement: ThreadDisengagement): readonly string[] { + const participants: string[] = [] + const add = (user: string | undefined): void => { + if (user === undefined || user.length === 0) return + if (user === disengagement.botUserId) return + if (!participants.includes(user)) participants.push(user) + } + for (const message of disengagement.messages) { + if (message.botId !== undefined) continue + add(message.user) + } + add(disengagement.replierUserId) + return participants +} + +/** + * Who actually receives the DM, which is deliberately not the same as who was + * in the thread. The two disengagement reasons describe opposite situations and + * a single fan-out rule cannot serve both. + * + * `thread-dormant` — the conversation itself went quiet for a day and the bot + * stepped out of it. Nobody is mid-discussion, everyone who took part has a + * stake in knowing it is no longer listening, and the DM is the only thing that + * will tell them. Everyone gets it. + * + * `engagement-buried` — the exact opposite: the thread is *lively*, it has + * simply run past the bot's last involvement. Fanning out here would DM people + * who moved on long ago, about a message they did not write, in a thread that + * is still busy — the notification becomes the noise. Only the person whose + * reply went unanswered actually needs to know. + * + * Falls back to the full list when the replier is unknown or is the bot itself: + * telling the wrong set of people is recoverable, telling nobody is the failure + * this whole path exists to prevent. + * + * Exported for the tests; `notifyThreadDisengagement` calls it itself. + */ +export function noticeRecipients(disengagement: ThreadDisengagement): readonly string[] { + const participants = humanThreadParticipants(disengagement) + if (disengagement.reason !== "engagement-buried") return participants + + const replier = disengagement.replierUserId + if (replier === undefined || replier.length === 0 || replier === disengagement.botUserId) { + return participants + } + return [replier] +} + +/** + * DMs that the bot did not pick up the latest reply, and how to bring it back. + * Who receives it depends on why it disengaged — see `noticeRecipients`. + * Intended to be fired without awaiting — it never throws. + */ +export async function notifyThreadDisengagement( + disengagement: ThreadDisengagement, + deps: DisengageNoticeDeps = defaultDisengageNoticeDeps, +): Promise { + try { + const key = noticeKey(disengagement.teamId, disengagement.channelId, disengagement.threadTs) + if (noticedThreads.get(key) !== undefined) return + + const recipients = noticeRecipients(disengagement) + if (recipients.length === 0) return + + // Claimed before the sends, not after: a Slack outage mid-notice must not + // leave the thread eligible to try the whole fan-out again on the next + // reply. One missed notice beats a repeating one. + noticedThreads.set(key, { expiresAt: Date.now() + NOTICE_TTL_MS }) + + const botToken = await deps.resolveBotToken({ teamId: disengagement.teamId }) + // One permalink for the whole fan-out — it is the same message. + const permalink = await deps + .getPermalink({ + botToken, + channelId: disengagement.channelId, + messageTs: disengagement.messageTs, + }) + .catch(() => null) + const text = disengagementNoticeText(disengagement.reason, permalink) + + // One recipient's closed DM (or a deactivated account) must not cost the + // others theirs. + await Promise.all( + recipients.map(async (userId) => { + try { + await deps.postDirectMessage({ botToken, userId, text }) + } catch (error) { + console.warn( + `[slack-disengage-notice] Failed to DM ${userId} about a thread the bot stopped following.`, + error, + ) + } + }), + ) + } catch (error) { + console.warn("[slack-disengage-notice] Disengagement notice failed — ignored.", error) + } +} + +/** + * The DM itself. Two things have to land: why the bot went quiet (so it does not + * read as a bug), and the one action that undoes it. The thread link comes + * second because the sentence has to make sense in a notification preview, + * where the link is just text. + * + * Exported for the tests. + */ +export function disengagementNoticeText(reason: FollowUpDisengagedReason, permalink: string | null): string { + const why = + reason === "thread-dormant" + ? "the thread had been quiet for over a day, so I'd stepped out of it" + : "the conversation had moved well past the last message I was part of, so I'd stepped out of it" + const where = permalink === null ? "a Slack thread" : `<${permalink}|a Slack thread>` + return [ + `:zzz: There's a new reply in ${where} I'd been working in, and I didn't pick it up — ${why}.`, + "", + "@-mention me in that thread and I'll jump straight back in.", + ].join("\n") +} + +/** + * `chat.getPermalink`. Returns null rather than throwing when Slack cannot + * produce one: the notice is still worth sending without a link. + * + * Exported for the request-shape tests; production reaches it through + * `defaultDisengageNoticeDeps`. + */ +export async function getPermalinkFromSlack(options: { + readonly botToken: string + readonly channelId: string + readonly messageTs: string +}): Promise { + // The one read method here, and Slack prefers GET for it. + const query = new URLSearchParams({ + channel: options.channelId, + message_ts: options.messageTs, + }) + const res = await fetch(`${SLACK_API}/chat.getPermalink?${query.toString()}`, { + headers: { authorization: `Bearer ${options.botToken}` }, + signal: AbortSignal.timeout(SLACK_API_TIMEOUT_MS), + }) + if (!res.ok) return null + const payload = (await res.json()) as { ok: boolean; permalink?: string } + if (!payload.ok || typeof payload.permalink !== "string") return null + return payload.permalink +} + +/** + * `chat.postMessage` addressed to a user id: Slack opens (or reuses) the DM + * conversation itself, so there is no `conversations.open` hop to make. + * + * Exported for the request-shape tests; production reaches it through + * `defaultDisengageNoticeDeps`. + */ +export async function postDirectMessageViaSlack(options: { + readonly botToken: string + readonly userId: string + readonly text: string +}): Promise { + const res = await fetch(`${SLACK_API}/chat.postMessage`, { + method: "POST", + headers: { + authorization: `Bearer ${options.botToken}`, + "content-type": "application/json; charset=utf-8", + }, + body: JSON.stringify({ channel: options.userId, text: options.text }), + signal: AbortSignal.timeout(SLACK_API_TIMEOUT_MS), + }) + if (!res.ok) { + throw new Error(`Slack chat.postMessage failed: HTTP ${res.status}`) + } + const payload = (await res.json()) as { ok: boolean; error?: string } + if (!payload.ok) { + throw new Error(`Slack chat.postMessage failed: ${payload.error ?? "unknown_error"}`) + } +} diff --git a/apps/slack-agent/agent/lib/thread-context.test.ts b/apps/slack-agent/agent/lib/thread-context.test.ts index 18a44496b..0b876e423 100644 --- a/apps/slack-agent/agent/lib/thread-context.test.ts +++ b/apps/slack-agent/agent/lib/thread-context.test.ts @@ -2,7 +2,7 @@ import { afterEach, describe, expect, test } from "bun:test" import type { SlackThreadMessage } from "eve/channels/slack" import { botUserIdForTeam, rememberBotUserId, resetBotUserIdCacheForTests } from "./bot-identity.js" import { slackMessageContent } from "./slack-context-format.js" -import { formatThreadContext, loadThreadContext } from "./thread-context.js" +import { formatThreadContext, loadThreadMessages } from "./thread-context.js" afterEach(() => { resetBotUserIdCacheForTests() @@ -206,7 +206,7 @@ describe("formatThreadContext", () => { }) }) -describe("loadThreadContext", () => { +describe("loadThreadMessages", () => { const thread = (messages: readonly SlackThreadMessage[]) => { const recentMessages: SlackThreadMessage[] = [] return { @@ -218,27 +218,31 @@ describe("loadThreadContext", () => { } } - test("includes the whole thread before the triggering message", async () => { + test("returns the whole thread before the triggering message", async () => { const trigger = userMessage("<@UBOT> why did this fire?", "1700000000.000300") - const context = await loadThreadContext( + const messages = await loadThreadMessages( thread([alertMessage(), userMessage("hm", "1700000000.000200"), trigger]), { threadTs: trigger.threadTs, ts: trigger.ts }, - { botUserId: BOT_USER_ID }, ) - expect(context).toHaveLength(1) + expect(messages?.map((message) => message.ts)).toEqual(["1700000000.000100", "1700000000.000200"]) + + const context = formatThreadContext(messages ?? [], { botUserId: BOT_USER_ID }) // The alert survives — this is the bug: `since: "last-agent-reply"` used to // cut context off after it, leaving the model nothing to answer from. - expect(context[0]).toContain("🚨 checkout p95 — Firing") - expect(context[0]).toContain("hm") - expect(context[0]).not.toContain("why did this fire?") + expect(context).toContain("🚨 checkout p95 — Firing") + expect(context).toContain("hm") + expect(context).not.toContain("why did this fire?") }) - test("adds nothing for a thread root", async () => { + test("returns nothing for a thread root", async () => { const root = userMessage("<@UBOT> hello", "1700000000.000100") - expect(await loadThreadContext(thread([root]), { threadTs: root.ts, ts: root.ts })).toEqual([]) + expect(await loadThreadMessages(thread([root]), { threadTs: root.ts, ts: root.ts })).toEqual([]) }) - test("degrades to no context when Slack fails, so the turn still dispatches", async () => { + test("returns null when Slack fails, which is not the same as an empty thread", async () => { + // The turn still dispatches without context; the follow-up check reads the + // difference too — "nobody said anything" declines, "we could not look" + // fails open (#lib/thread-follow-up.js). const failing = { recentMessages: [] as SlackThreadMessage[], refresh: async () => { @@ -246,8 +250,8 @@ describe("loadThreadContext", () => { }, } expect( - await loadThreadContext(failing, { threadTs: "1700000000.000100", ts: "1700000000.000300" }), - ).toEqual([]) + await loadThreadMessages(failing, { threadTs: "1700000000.000100", ts: "1700000000.000300" }), + ).toBeNull() }) }) diff --git a/apps/slack-agent/agent/lib/thread-context.ts b/apps/slack-agent/agent/lib/thread-context.ts index f790ae580..99e0d4505 100644 --- a/apps/slack-agent/agent/lib/thread-context.ts +++ b/apps/slack-agent/agent/lib/thread-context.ts @@ -39,32 +39,36 @@ import { formatContextBlock, type ContextFormatOptions } from "./slack-context-f * thread pays for its history once per mention), and eve's `thread.refresh()` * fetches one oldest-first page of 50 replies, so past 50 the tail is the part * that goes missing. Alert threads are short; revisit if that stops holding. + * + * Loading and rendering are separate steps rather than one `loadThreadContext` + * call because this transcript now has a second reader: `confirmThreadFollowUp` + * (`#lib/thread-follow-up.js`) decides from these same messages whether an + * optimistically promoted follow-up becomes a turn at all. Fetching the thread + * twice for two questions about the same thread would be the easy mistake here + * — and one of them would be answering about a slightly different thread. */ export type ThreadContextOptions = ContextFormatOptions /** - * Loads the thread the triggering message belongs to and renders it as one - * context string, or `[]` when there is nothing to add (thread root, empty - * thread, or a Slack call that failed). + * Loads the thread the triggering message belongs to, oldest-first and without + * the triggering message itself. Returns `[]` for a thread root (there is + * nothing before it) and `null` when Slack could not be read at all — callers + * need those apart: "empty thread" is an answer, "no answer" is not. * * Never throws: eve drops the whole mention when an `onAppMention` handler * throws, and losing the reply entirely is far worse than losing its context. */ -export async function loadThreadContext( +export async function loadThreadMessages( thread: Pick, message: { readonly threadTs: string; readonly ts: string }, - options: ThreadContextOptions = {}, -): Promise { - let messages: readonly SlackThreadMessage[] +): Promise { try { - messages = await loadThreadContextMessages(thread, message, { since: "thread-root" }) + return await loadThreadContextMessages(thread, message, { since: "thread-root" }) } catch (error) { console.warn("[slack-thread-context] Failed to load thread context; dispatching without it.", error) - return [] + return null } - const rendered = formatThreadContext(messages, options) - return rendered === undefined ? [] : [rendered] } /** diff --git a/apps/slack-agent/agent/lib/thread-follow-up.test.ts b/apps/slack-agent/agent/lib/thread-follow-up.test.ts index 5dffd8fe9..e9732f73b 100644 --- a/apps/slack-agent/agent/lib/thread-follow-up.test.ts +++ b/apps/slack-agent/agent/lib/thread-follow-up.test.ts @@ -1,15 +1,19 @@ import { afterEach, beforeEach, describe, expect, setSystemTime, test } from "bun:test" +import type { SlackThreadMessage } from "eve/channels/slack" import { installFetchStub } from "./fetch-stub.js" import { - fetchThreadRepliesFromSlack, + confirmThreadFollowUp, + pendingFollowUp, promoteThreadFollowUp, recordThreadEngagement, - resetThreadEngagementCacheForTests, - type ThreadFollowUpDeps, - type ThreadReplyMessage, + resetThreadFollowUpStateForTests, + type PendingFollowUp, } from "./thread-follow-up.js" const BOT_USER_ID = "U0BOT" +const TEAM_ID = "T123" +const CHANNEL_ID = "C123" +const THREAD_TS = "1700000000.000100" // ── helpers ───────────────────────────────────────────────────────────────── @@ -19,74 +23,88 @@ function envelope(overrides: { }): string { return JSON.stringify({ type: "event_callback", - team_id: "T123", + team_id: TEAM_ID, event_id: "Ev123", authorizations: [{ user_id: BOT_USER_ID, is_bot: true }], event: { type: "message", channel_type: "channel", - channel: "C123", + channel: CHANNEL_ID, user: "U456", text: "can you investigate the root cause?", ts: "1700000002.000200", - thread_ts: "1700000000.000100", + thread_ts: THREAD_TS, ...overrides.event, }, ...overrides.envelope, }) } -function makeDeps(replies: readonly ThreadReplyMessage[]): { - deps: ThreadFollowUpDeps - calls: () => number -} { - let fetchCalls = 0 - return { - deps: { - resolveBotToken: async () => "xoxb-test", - fetchThreadReplies: async () => { - fetchCalls += 1 - return replies - }, - }, - calls: () => fetchCalls, +/** + * Promotes a body and hands back the pending registration the handler would + * later look up — the two halves are only ever used together. + */ +function promote( + body: string = envelope({}), + options: { readonly isSlackRetry?: boolean } = {}, +): PendingFollowUp | null { + const promoted = promoteThreadFollowUp(body, options) + if (promoted === null) return null + const parsed = JSON.parse(promoted) as { + team_id?: string + event: { channel: string; ts: string } } + return pendingFollowUp({ + teamId: parsed.team_id, + channelId: parsed.event.channel, + messageTs: parsed.event.ts, + }) } -// The default envelope's reply is at ts 1700000002.000200, so both of these -// engagements are seconds old — comfortably inside the recency bound. -const ENGAGED_THREAD: readonly ThreadReplyMessage[] = [ - { - user: "U456", - text: `<@${BOT_USER_ID}> why is error rate up?`, - ts: "1700000000.000100", - }, - { - user: BOT_USER_ID, - text: "Here are the reasons why...", - ts: "1700000001.000100", - }, +const threadMessage = (overrides: Partial = {}): SlackThreadMessage => ({ + text: "", + markdown: "", + user: undefined, + botId: undefined, + ts: THREAD_TS, + threadTs: THREAD_TS, + isMe: false, + raw: {}, + ...overrides, +}) + +const humanMessage = (text: string, ts: string, user = "U456"): SlackThreadMessage => + threadMessage({ text, markdown: text, user, ts }) + +/** The bot's own reply as eve returns it: `user` is the bot AND `botId` is set. */ +const botMessage = (ts: string, text = "Here are the reasons why..."): SlackThreadMessage => + threadMessage({ text, markdown: text, user: BOT_USER_ID, botId: "B0BOT", ts }) + +// The default envelope's reply is at ts 1700000002.000200, so this thread is +// alive and the bot is one message behind it. +const ENGAGED_THREAD: readonly SlackThreadMessage[] = [ + humanMessage(`<@${BOT_USER_ID}> why is error rate up?`, "1700000000.000100"), + botMessage("1700000001.000100"), ] -const UNRELATED_THREAD: readonly ThreadReplyMessage[] = [ - { user: "U456", text: "lunch?", ts: "1700000000.000100" }, - { user: "U789", text: "sure", ts: "1700000001.000100" }, +const UNRELATED_THREAD: readonly SlackThreadMessage[] = [ + humanMessage("lunch?", "1700000000.000100"), + humanMessage("sure", "1700000001.000100", "U789"), ] beforeEach(() => { - resetThreadEngagementCacheForTests() + resetThreadFollowUpStateForTests() }) afterEach(() => { setSystemTime() // restore the real clock }) -// ── promotion ─────────────────────────────────────────────────────────────── +// ── promotion (webhook path) ──────────────────────────────────────────────── describe("promoteThreadFollowUp", () => { - test("promotes a follow-up reply in an engaged thread to app_mention", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const promoted = await promoteThreadFollowUp(envelope({}), deps) + test("promotes a follow-up reply in a channel thread to app_mention", () => { + const promoted = promoteThreadFollowUp(envelope({})) expect(promoted).not.toBeNull() const parsed = JSON.parse(promoted!) as { event: Record @@ -94,354 +112,359 @@ describe("promoteThreadFollowUp", () => { } expect(parsed.event.type).toBe("app_mention") // Everything else is preserved so eve's parser sees a coherent event. - expect(parsed.event.channel).toBe("C123") - expect(parsed.event.thread_ts).toBe("1700000000.000100") + expect(parsed.event.channel).toBe(CHANNEL_ID) + expect(parsed.event.thread_ts).toBe(THREAD_TS) expect(parsed.event.user).toBe("U456") expect(parsed.event_id).toBe("Ev123") }) - test("promotes when the bot was mentioned in the thread but has not replied yet", async () => { - const { deps } = makeDeps([ - { - user: "U456", - text: `<@${BOT_USER_ID}> why is error rate up?`, - ts: "1700000000.000100", - }, - ]) - expect(await promoteThreadFollowUp(envelope({}), deps)).not.toBeNull() + test("is synchronous and spends no network at all", () => { + // Incident 1: promotion used to resolve the workspace and fetch the thread + // INSIDE Slack's ~3s webhook budget, and a cold cache after a deploy blew + // the 2s deadline — 2009ms, no turn, no log, no ack. There is no deadline + // to miss any more because there is nothing here to wait for. + const stub = installFetchStub(() => { + throw new Error("promotion must not touch the network") + }) + try { + const promoted = promoteThreadFollowUp(envelope({})) + // Not a promise: a `then` here would mean something can still be awaited. + expect(typeof promoted).toBe("string") + expect(stub.calls.length).toBe(0) + } finally { + stub.restore() + } }) - test("private-channel (group) replies qualify too", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const body = envelope({ event: { channel_type: "group" } }) - expect(await promoteThreadFollowUp(body, deps)).not.toBeNull() + test("registers the reply as pending, keyed by team + channel + message ts", () => { + const pending = promote() + expect(pending?.teamId).toBe(TEAM_ID) + expect(pending?.channelId).toBe(CHANNEL_ID) + expect(pending?.threadTs).toBe(THREAD_TS) + expect(pending?.messageTs).toBe("1700000002.000200") + // Carried from the envelope's `authorizations`, so the confirmation never + // has to guess which user id is ours. + expect(pending?.botUserId).toBe(BOT_USER_ID) + expect(pending?.ackable).toBe(true) }) - test("group-DM (mpim) replies qualify too", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const body = envelope({ event: { channel_type: "mpim" } }) - expect(await promoteThreadFollowUp(body, deps)).not.toBeNull() + test("a Slack redelivery is still promoted, but does not re-ack", () => { + // The :eyes: is already on the message; `already_reacted` is tolerated + // downstream, this just skips the pointless call. + expect(promote(envelope({}), { isSlackRetry: true })?.ackable).toBe(false) }) - test("file_share subtype replies qualify (mirrors eve's DM filter)", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const body = envelope({ event: { subtype: "file_share" } }) - expect(await promoteThreadFollowUp(body, deps)).not.toBeNull() + test("private-channel (group) replies qualify too", () => { + expect(promoteThreadFollowUp(envelope({ event: { channel_type: "group" } }))).not.toBeNull() }) -}) -// ── engagement recency bounds ─────────────────────────────────────────────── - -// Engagement is not permanent: once anyone has mentioned the bot, an unbounded -// rule would dispatch a full agent turn for every later human reply in that -// thread forever — a cost amplifier and a standing prompt-injection intake. - -describe("engagement recency", () => { - test("a stale engagement (>30 min before the reply) does not promote", async () => { - const { deps } = makeDeps([ - { - user: BOT_USER_ID, - text: "Here are the reasons why...", - ts: "1700000001.000100", - }, - ]) - // 31 minutes after the bot's last message in the thread. - const body = envelope({ event: { ts: "1700001861.000200" } }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() - }) - - test("an engagement just inside the 30-minute window still promotes", async () => { - const { deps } = makeDeps([ - { - user: BOT_USER_ID, - text: "Here are the reasons why...", - ts: "1700000001.000100", - }, - ]) - const body = envelope({ event: { ts: "1700001799.000200" } }) - expect(await promoteThreadFollowUp(body, deps)).not.toBeNull() - }) - - test("an engagement pushed out of the trailing message window does not promote", async () => { - // Bot spoke first, then 20 human messages buried it — recent in time, but - // the conversation has demonstrably moved on. - const chatter: ThreadReplyMessage[] = Array.from({ length: 20 }, (_, i) => ({ - user: "U456", - text: `chatter ${i}`, - ts: `17000000${String(10 + i).padStart(2, "0")}.000100`, - })) - const { deps } = makeDeps([{ user: BOT_USER_ID, text: "on it", ts: "1700000001.000100" }, ...chatter]) - const body = envelope({ event: { ts: "1700000031.000200" } }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() + test("group-DM (mpim) replies qualify too", () => { + expect(promoteThreadFollowUp(envelope({ event: { channel_type: "mpim" } }))).not.toBeNull() }) - test("a message with no ts cannot be aged, so it does not count as engagement", async () => { - const { deps } = makeDeps([{ user: BOT_USER_ID, text: "on it" }]) - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() + test("file_share subtype replies qualify (mirrors eve's DM filter)", () => { + expect(promoteThreadFollowUp(envelope({ event: { subtype: "file_share" } }))).not.toBeNull() }) }) // ── pass-through cases ────────────────────────────────────────────────────── describe("pass-through", () => { - test("thread the bot is not part of", async () => { - const { deps } = makeDeps(UNRELATED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() + const passesThrough = (body: string): void => { + expect(promoteThreadFollowUp(body)).toBeNull() + // Nothing registered either: the handler must treat these as real events. + expect( + pendingFollowUp({ teamId: TEAM_ID, channelId: CHANNEL_ID, messageTs: "1700000002.000200" }), + ).toBeNull() + } + + test("reply that already @-mentions the bot (arrives as a real app_mention)", () => { + passesThrough(envelope({ event: { text: `<@${BOT_USER_ID}> and what about latency?` } })) }) - test("reply that already @-mentions the bot (arrives as a real app_mention)", async () => { - const { deps, calls } = makeDeps(ENGAGED_THREAD) - const body = envelope({ - event: { text: `<@${BOT_USER_ID}> and what about latency?` }, - }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() - // Rejected before any Slack API call. - expect(calls()).toBe(0) + test("bot-authored replies (no self-triggering loop)", () => { + passesThrough(envelope({ event: { bot_id: "B999" } })) }) - test("bot-authored replies (no self-triggering loop)", async () => { - const { deps, calls } = makeDeps(ENGAGED_THREAD) - const body = envelope({ event: { bot_id: "B999" } }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() - expect(calls()).toBe(0) + test("top-level channel messages (not a thread reply)", () => { + passesThrough(envelope({ event: { thread_ts: undefined } })) + expect(promoteThreadFollowUp(envelope({ event: { ts: THREAD_TS, thread_ts: THREAD_TS } }))).toBeNull() }) - test("top-level channel messages (not a thread reply)", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const noThread = envelope({ event: { thread_ts: undefined } }) - expect(await promoteThreadFollowUp(noThread, deps)).toBeNull() - const rootOfThread = envelope({ - event: { ts: "1700000000.000100", thread_ts: "1700000000.000100" }, - }) - expect(await promoteThreadFollowUp(rootOfThread, deps)).toBeNull() + test("DMs (eve dispatches those on its own)", () => { + passesThrough(envelope({ event: { channel_type: "im" } })) + }) + + test("edits and other subtypes", () => { + passesThrough(envelope({ event: { subtype: "message_changed" } })) }) - test("DMs (eve dispatches those on its own)", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const body = envelope({ event: { channel_type: "im" } }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() + test("non-message events", () => { + passesThrough(envelope({ event: { type: "reaction_added" } })) }) - test("edits and other subtypes", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const body = envelope({ event: { subtype: "message_changed" } }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() + test("envelope without authorizations (bot user unknown)", () => { + passesThrough(envelope({ envelope: { authorizations: undefined } })) }) - test("non-message events", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const body = envelope({ event: { type: "reaction_added" } }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() + test("interaction form posts / non-JSON bodies", () => { + passesThrough("payload=%7B%7D") }) - test("envelope without authorizations (bot user unknown)", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const body = envelope({ envelope: { authorizations: undefined } }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() + test("url_verification and other envelope types", () => { + passesThrough(JSON.stringify({ type: "url_verification", challenge: "x" })) + }) +}) + +describe("pendingFollowUp", () => { + test("a real app_mention was never promoted, so it has no pending entry", () => { + expect( + pendingFollowUp({ teamId: TEAM_ID, channelId: CHANNEL_ID, messageTs: "1700000002.000200" }), + ).toBeNull() }) - test("interaction form posts / non-JSON bodies", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - expect(await promoteThreadFollowUp("payload=%7B%7D", deps)).toBeNull() + test("the key includes the team: one workspace cannot answer for another", () => { + // Slack channel ids are only unique per workspace, and this registry is + // process-global across every tenant that installed the app. + promote() + expect( + pendingFollowUp({ teamId: "T999", channelId: CHANNEL_ID, messageTs: "1700000002.000200" }), + ).toBeNull() }) - test("url_verification and other envelope types", async () => { - const { deps } = makeDeps(ENGAGED_THREAD) - const body = JSON.stringify({ type: "url_verification", challenge: "x" }) - expect(await promoteThreadFollowUp(body, deps)).toBeNull() + test("reading it does not consume it: the same message is judged the same way twice", () => { + promote() + const target = { teamId: TEAM_ID, channelId: CHANNEL_ID, messageTs: "1700000002.000200" } + expect(pendingFollowUp(target)).not.toBeNull() + expect(pendingFollowUp(target)).not.toBeNull() }) }) -// ── fetch bounds ──────────────────────────────────────────────────────────── - -// `conversations.replies` returns the OLDEST page first. Without oldest/latest -// bounds, a thread past 100 replies has its "trailing window" computed over the -// oldest page: fresh engagements invisible, stale ones passing recency. - -describe("thread fetch bounds", () => { - test("the fetch is bounded to the recency horizon ending at the reply", async () => { - let received: Parameters[0] | undefined - const deps: ThreadFollowUpDeps = { - resolveBotToken: async () => "xoxb-test", - fetchThreadReplies: async (options) => { - received = options - return ENGAGED_THREAD - }, - } - expect(await promoteThreadFollowUp(envelope({}), deps)).not.toBeNull() - // 30 minutes (ENGAGEMENT_MAX_AGE_SECONDS) before the incoming reply's ts. - expect(received?.oldest).toBe(String(Number("1700000002.000200") - 30 * 60)) - expect(received?.latest).toBe("1700000002.000200") - expect(received?.signal).toBeInstanceOf(AbortSignal) +// ── confirmation (dispatch path) ──────────────────────────────────────────── + +describe("confirmThreadFollowUp", () => { + test("promotes a reply in a thread the bot replied in", () => { + expect(confirmThreadFollowUp(promote()!, ENGAGED_THREAD)).toEqual({ + engaged: true, + reason: "recent-engagement", + }) + }) + + test("promotes when the bot was mentioned in the thread but has not replied yet", () => { + const mentionOnly = [humanMessage(`<@${BOT_USER_ID}> why is error rate up?`, THREAD_TS)] + expect(confirmThreadFollowUp(promote()!, mentionOnly).engaged).toBe(true) }) - test("the Slack request itself carries oldest/latest/inclusive", async () => { - const stub = installFetchStub(() => Response.json({ ok: true, messages: [] })) + test("spends no network: it reads the thread the handler already loaded", () => { + const stub = installFetchStub(() => { + throw new Error("confirmation must not fetch the thread a second time") + }) try { - await fetchThreadRepliesFromSlack({ - botToken: "xoxb-test", - channelId: "C123", - threadTs: "1700000000.000100", - oldest: "1699998202.0002", - latest: "1700000002.000200", - signal: AbortSignal.timeout(1_000), - }) - expect(stub.calls.length).toBe(1) - const params = new URLSearchParams(String(stub.calls[0]?.body)) - expect(params.get("channel")).toBe("C123") - expect(params.get("ts")).toBe("1700000000.000100") - expect(params.get("oldest")).toBe("1699998202.0002") - expect(params.get("latest")).toBe("1700000002.000200") - expect(params.get("inclusive")).toBe("true") + expect(confirmThreadFollowUp(promote()!, ENGAGED_THREAD).engaged).toBe(true) + expect(stub.calls.length).toBe(0) } finally { stub.restore() } }) + + test("a thread nobody has touched for over a day is dormant", () => { + // 24h + 1 minute after the previous message in the thread. + const pending = promote(envelope({ event: { ts: "1700086461.000200" } }))! + expect(confirmThreadFollowUp(pending, ENGAGED_THREAD)).toEqual({ + engaged: false, + workedInThread: true, + reason: "thread-dormant", + }) + }) + + test("dormancy is measured from the thread's last message, not the bot's", () => { + // THE REGRESSION. A user replied 61.3 minutes after the bot's last post in + // a thread people were still using. The old rule expired engagement after + // 30 minutes AND derived the `conversations.replies` window from that same + // constant — so the fetch began 31 minutes after the bot had last spoken + // and the check could not see the bot's own messages at all. It answered + // "not engaged" in 406ms: no timeout, just confidently wrong. + const botPostSeconds = 1700000001 + const replySeconds = botPostSeconds + Math.round(61.3 * 60) + const thread = [ + humanMessage(`<@${BOT_USER_ID}> why is error rate up?`, "1700000000.000100"), + botMessage(`${botPostSeconds}.000100`), + // The thread kept moving while the bot stayed quiet — which is exactly + // what a live incident thread looks like. + humanMessage("checking the deploy log", `${botPostSeconds + 1800}.000100`, "U789"), + ] + const pending = promote(envelope({ event: { ts: `${replySeconds}.000200` } }))! + expect(confirmThreadFollowUp(pending, thread)).toEqual({ + engaged: true, + reason: "recent-engagement", + }) + }) + + test("an engagement pushed out of the trailing message window does not promote", () => { + // Bot spoke first, then 14 human messages buried it — the conversation has + // demonstrably moved on, however recently it did so. + const chatter = Array.from({ length: 14 }, (_, i) => + humanMessage(`chatter ${i}`, `1700000${String(10 + i).padStart(3, "0")}.000100`), + ) + const pending = promote(envelope({ event: { ts: "1700000100.000200" } }))! + expect(confirmThreadFollowUp(pending, [botMessage("1700000001.000100"), ...chatter])).toEqual({ + engaged: false, + workedInThread: true, + reason: "engagement-buried", + }) + }) + + test("an engagement still inside the trailing window promotes", () => { + // One fewer message: the bot, 13 replies and this one make 15. + const chatter = Array.from({ length: 13 }, (_, i) => + humanMessage(`chatter ${i}`, `1700000${String(10 + i).padStart(3, "0")}.000100`), + ) + const pending = promote(envelope({ event: { ts: "1700000100.000200" } }))! + expect(confirmThreadFollowUp(pending, [botMessage("1700000001.000100"), ...chatter]).engaged).toBe( + true, + ) + }) + + test("a thread the bot has never been in is declined, and stays silent", () => { + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD)).toEqual({ + engaged: false, + workedInThread: false, + reason: "never-engaged", + }) + }) + + test("another bot's post is not our engagement", () => { + // eve's `isMe` is `bot_id !== undefined`, i.e. ANY bot — so a GitHub or CI + // app in the same channel would otherwise read as us. + const github = threadMessage({ + text: "PR #338 merged", + markdown: "PR #338 merged", + user: "U0GITHUB", + botId: "B0GITHUB", + isMe: true, + ts: "1700000001.000100", + }) + expect(confirmThreadFollowUp(promote()!, [github]).engaged).toBe(false) + }) + + test("a Maple alert card counts: the bot posted it, and the thread hangs off it", () => { + const alert = threadMessage({ + user: BOT_USER_ID, + botId: "B0BOT", + isMe: true, + ts: "1700000001.000100", + raw: { attachments: [{ fallback: "🚨 checkout p95 — Firing" }] }, + }) + expect(confirmThreadFollowUp(promote()!, [alert]).engaged).toBe(true) + }) + + test("mentions are matched on the raw text, which eve's markdown may have rewritten", () => { + const mention = threadMessage({ + text: `<@${BOT_USER_ID}> take a look`, + markdown: "@maple take a look", + user: "U789", + ts: "1700000001.000100", + }) + expect(confirmThreadFollowUp(promote()!, [mention]).engaged).toBe(true) + }) }) -// ── promotion deadline ────────────────────────────────────────────────────── - -// The whole promotion side-trip (workspace resolve + thread fetch) shares one -// deadline inside Slack's ~3s webhook budget; expiry falls through unpromoted. - -describe("promotion deadline", () => { - test("a slow workspace resolve falls through unpromoted at the deadline", async () => { - let fetchCalls = 0 - const deps: ThreadFollowUpDeps = { - resolveBotToken: () => new Promise((resolve) => setTimeout(() => resolve("xoxb-test"), 200)), - fetchThreadReplies: async () => { - fetchCalls += 1 - return ENGAGED_THREAD - }, - promotionDeadlineMs: 20, - } - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() - expect(fetchCalls).toBe(0) +// ── failing open ──────────────────────────────────────────────────────────── - // Expiry cached nothing: the thread's next reply retries and promotes. - const { deps: freshDeps } = makeDeps(ENGAGED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), freshDeps)).not.toBeNull() +// A wrong drop costs a user their message with nothing on screen to explain it; +// a wrong dispatch costs one turn in a thread the bot was already part of. The +// two are not symmetric, so absent evidence promotes. + +describe("failing open", () => { + test("an unreadable thread promotes rather than losing the message", () => { + expect(confirmThreadFollowUp(promote()!, null)).toEqual({ + engaged: true, + reason: "thread-unreadable", + }) }) - test("a slow thread fetch falls through unpromoted at the deadline", async () => { - const deps: ThreadFollowUpDeps = { - resolveBotToken: async () => "xoxb-test", - fetchThreadReplies: () => - new Promise((resolve) => setTimeout(() => resolve(ENGAGED_THREAD), 200)), - promotionDeadlineMs: 20, - } - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() - }) - - test("expiry is logged: it is the one path that drops a message with no other trace", async () => { - // No turn is created, eve drops the event as `unsupported` without a line - // of its own, and the :eyes: ack never fires — so unless this logs, the - // bot going quiet leaves nothing behind to find. hooks/outcome-log.ts - // cannot cover it: there is no turn for it to report on. - const warnings: string[] = [] - const realWarn = console.warn - console.warn = (line: unknown) => { - warnings.push(String(line)) - } - try { - const deps: ThreadFollowUpDeps = { - resolveBotToken: async () => "xoxb-test", - fetchThreadReplies: () => - new Promise((resolve) => setTimeout(() => resolve(ENGAGED_THREAD), 200)), - promotionDeadlineMs: 20, - } - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() - } finally { - console.warn = realWarn - } + test("a truncated page with no visible engagement promotes", () => { + // eve's `thread.refresh()` fetches ONE oldest-first page of 50, so on a + // longer thread the tail — where any recent engagement lives — is exactly + // what is missing. + const longThread = Array.from({ length: 49 }, (_, i) => + humanMessage(`chatter ${i}`, `1700000${String(100 + i).padStart(3, "0")}.000100`), + ) + const pending = promote(envelope({ event: { ts: "1700000200.000200" } }))! + expect(confirmThreadFollowUp(pending, longThread)).toEqual({ + engaged: true, + reason: "page-truncated", + }) + }) - const logged = warnings.map((line) => JSON.parse(line) as Record) - const timeout = logged.find( - (entry) => entry["maple.agent.event"] === "follow_up_promotion_timeout", + test("a short page with no engagement is a real answer, not a missing one", () => { + const shortThread = Array.from({ length: 48 }, (_, i) => + humanMessage(`chatter ${i}`, `1700000${String(100 + i).padStart(3, "0")}.000100`), ) - expect(timeout).toBeDefined() - // Enough to find the thread in Slack; never the message text. - expect(timeout?.["maple.slack.team_id"]).toBe("T123") - expect(timeout?.["maple.slack.channel_id"]).toBe("C123") - expect(timeout?.["maple.slack.thread_ts"]).toBe("1700000000.000100") - expect(timeout?.["maple.agent.deadline_ms"]).toBe(20) + const pending = promote(envelope({ event: { ts: "1700000200.000200" } }))! + expect(confirmThreadFollowUp(pending, shortThread).engaged).toBe(false) }) }) // ── caching ───────────────────────────────────────────────────────────────── describe("engagement cache", () => { - test("second follow-up in the same thread skips the Slack API call", async () => { - const { deps, calls } = makeDeps(ENGAGED_THREAD) - await promoteThreadFollowUp(envelope({}), deps) - const second = envelope({ event: { ts: "1700000003.000300" } }) - expect(await promoteThreadFollowUp(second, deps)).not.toBeNull() - expect(calls()).toBe(1) - }) - - test("threads are cached independently", async () => { - const { deps, calls } = makeDeps(ENGAGED_THREAD) - await promoteThreadFollowUp(envelope({}), deps) - const otherThread = envelope({ - event: { thread_ts: "1700000010.000100", ts: "1700000011.000200" }, + test("a warm entry short-circuits the whole check", () => { + recordThreadEngagement(botPost()) + // The page says otherwise and loses: the cache is proof of something the + // bot did here within the last five minutes, which the page (one oldest- + // first slice of a possibly much longer thread) cannot contradict. + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD)).toEqual({ + engaged: true, + reason: "cached-engagement", }) - await promoteThreadFollowUp(otherThread, deps) - expect(calls()).toBe(2) }) - test("a second NON-engaged reply is served from the negative cache", async () => { - const { deps, calls } = makeDeps(UNRELATED_THREAD) - setSystemTime(new Date("2026-07-21T12:00:00Z")) - - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() - expect(calls()).toBe(1) + test("a confirmation from the page warms the cache for the next follow-up", () => { + expect(confirmThreadFollowUp(promote()!, ENGAGED_THREAD).reason).toBe("recent-engagement") + const second = promote(envelope({ event: { ts: "1700000003.000300" } }))! + expect(confirmThreadFollowUp(second, UNRELATED_THREAD).reason).toBe("cached-engagement") + }) - // Still inside the 20s negative TTL: no second round-trip to Slack, which - // is what keeps a busy channel the bot was never in from costing an API - // call per message. - setSystemTime(new Date("2026-07-21T12:00:19Z")) - const second = envelope({ event: { ts: "1700000003.000300" } }) - expect(await promoteThreadFollowUp(second, deps)).toBeNull() - expect(calls()).toBe(1) + test("a declined confirmation caches nothing", () => { + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(false) + // No negative entry to go stale: the bot joining the thread a second later + // is picked up by the very next reply. + const second = promote(envelope({ event: { ts: "1700000003.000300" } }))! + expect(confirmThreadFollowUp(second, ENGAGED_THREAD).engaged).toBe(true) }) - test("the negative cache expires so a thread the bot just joined is picked up", async () => { - const { deps, calls } = makeDeps(UNRELATED_THREAD) + test("the entry expires, so a thread that went quiet is re-judged from the page", () => { setSystemTime(new Date("2026-07-21T12:00:00Z")) - await promoteThreadFollowUp(envelope({}), deps) + recordThreadEngagement(botPost()) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(true) - setSystemTime(new Date("2026-07-21T12:00:21Z")) - await promoteThreadFollowUp(envelope({ event: { ts: "1700000004.000400" } }), deps) - expect(calls()).toBe(2) + // Past ENGAGED_TTL_MS: the cache no longer proves anything and the thread + // page has the only say. + setSystemTime(new Date("2026-07-21T12:05:01Z")) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(false) }) - test("the cache key includes the team: one workspace cannot answer for another", async () => { - // Slack channel ids are only unique per workspace, and this cache is - // process-global across every tenant that installed the app. - let call = 0 - const deps = { - resolveBotToken: async () => "xoxb-test", - fetchThreadReplies: async () => { - call += 1 - return call === 1 ? ENGAGED_THREAD : UNRELATED_THREAD - }, - } + test("threads are cached independently", () => { + recordThreadEngagement(botPost()) + const otherThread = promote( + envelope({ event: { thread_ts: "1700000010.000100", ts: "1700000011.000200" } }), + )! + expect(confirmThreadFollowUp(otherThread, UNRELATED_THREAD).engaged).toBe(false) + }) - expect(await promoteThreadFollowUp(envelope({}), deps)).not.toBeNull() - // Same channel + thread ids, different workspace: must be resolved on its - // own, and must not inherit the first team's engagement. - const otherTeam = envelope({ envelope: { team_id: "T999" } }) - expect(await promoteThreadFollowUp(otherTeam, deps)).toBeNull() - expect(call).toBe(2) + test("the cache key includes the team: one workspace cannot answer for another", () => { + recordThreadEngagement(botPost()) + const otherTeam = promote(envelope({ envelope: { team_id: "T999" } }))! + expect(confirmThreadFollowUp(otherTeam, UNRELATED_THREAD).engaged).toBe(false) }) }) // ── engagement learned from the event stream ──────────────────────────────── -// The cold path (conversations.replies) is the one that has to fit inside -// Slack's webhook budget, and the one that drops a user's message when it -// doesn't. Everything the bot does in a thread already comes back through the -// events stream, so most threads should never reach it at all. +// Everything the bot does in a thread already comes back through the events +// stream, so the cheap path should carry most threads — and, unlike the thread +// page, it still works past eve's 50-message limit. /** The bot's own post echoing back: `bot_id` set AND `user` = the bot. */ function botPost(overrides: Record = {}): string { @@ -457,28 +480,21 @@ function botPost(overrides: Record = {}): string { } describe("recordThreadEngagement", () => { - test("the bot's own post warms the thread, so the next follow-up needs no fetch", async () => { + test("the bot's own post warms the thread", () => { recordThreadEngagement(botPost()) - - const { deps, calls } = makeDeps(UNRELATED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), deps)).not.toBeNull() - // Promoted purely from what the event stream already told us. - expect(calls()).toBe(0) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(true) }) - test("an @mention of the bot warms the thread too", async () => { + test("an @mention of the bot warms the thread too", () => { recordThreadEngagement( envelope({ event: { text: `<@${BOT_USER_ID}> why is error rate up?`, ts: "1700000001.000100" }, }), ) - - const { deps, calls } = makeDeps(UNRELATED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), deps)).not.toBeNull() - expect(calls()).toBe(0) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(true) }) - test("a root-level mention is keyed by its own ts — the ts that becomes the thread", async () => { + test("a root-level mention is keyed by its own ts — the ts that becomes the thread", () => { // No thread_ts yet: the thread does not exist until the bot replies. recordThreadEngagement( envelope({ @@ -487,99 +503,70 @@ describe("recordThreadEngagement", () => { channel_type: undefined, thread_ts: undefined, text: `<@${BOT_USER_ID}> take a look`, - ts: "1700000000.000100", + ts: THREAD_TS, }, }), ) - // The default envelope replies in thread 1700000000.000100 — the mention's // own ts. Its very first follow-up is already warm. - const { deps, calls } = makeDeps(UNRELATED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), deps)).not.toBeNull() - expect(calls()).toBe(0) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(true) }) - test("the regression: a follow-up 3 minutes after the bot's last post stays warm", async () => { - // The incident. Engagement had only ever been written by the cold path, so - // it aged out 5 minutes after the last *promotion* rather than after the - // bot's last post — and the next follow-up paid the full round-trip inside - // Slack's webhook budget, missed the deadline, and was dropped in silence. + test("the regression: a follow-up 3 minutes after the bot's last post, on a cold cache", () => { + // Incident 1, from the other side. A deploy wipes this cache, so the very + // next follow-up used to pay the full workspace-resolve + thread-fetch + // round-trip inside Slack's webhook budget, miss the 2s deadline and be + // dropped in silence. Nothing is warm here and nothing needs to be: the + // decision reads the thread the handler loaded anyway, off the budget. setSystemTime(new Date("2026-08-05T19:27:44Z")) - recordThreadEngagement(botPost({ ts: "1785958064.000100" })) + const botPostTs = "1785958064.000100" + const followUpTs = "1785958238.000200" // 2m54s later setSystemTime(new Date("2026-08-05T19:30:38Z")) - const { deps, calls } = makeDeps(UNRELATED_THREAD) - const followUp = envelope({ event: { ts: "1785958238.000200" } }) - expect(await promoteThreadFollowUp(followUp, deps)).not.toBeNull() - expect(calls()).toBe(0) + const stub = installFetchStub(() => { + throw new Error("a cold cache must not cost a network call any more") + }) + try { + const pending = promote(envelope({ event: { ts: followUpTs } }))! + const decision = confirmThreadFollowUp(pending, [ + humanMessage(`<@${BOT_USER_ID}> why is error rate up?`, THREAD_TS), + botMessage(botPostTs), + ]) + expect(decision).toEqual({ engaged: true, reason: "recent-engagement" }) + expect(stub.calls.length).toBe(0) + } finally { + stub.restore() + } }) - test("a stranger's message teaches nothing", async () => { + test("a stranger's message teaches nothing", () => { recordThreadEngagement(envelope({ event: { text: "lunch?" } })) - - const { deps, calls } = makeDeps(UNRELATED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() - // Fell through to the cold path rather than caching a false positive. - expect(calls()).toBe(1) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(false) }) - test("another bot's post teaches nothing (it is not our engagement)", async () => { + test("another bot's post teaches nothing (it is not our engagement)", () => { recordThreadEngagement(botPost({ user: "U0OTHERBOT", bot_id: "B0GITHUB" })) - - const { deps, calls } = makeDeps(UNRELATED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() - expect(calls()).toBe(1) - }) - - test("engagement never moves backwards when Slack redelivers an older event", async () => { - recordThreadEngagement(botPost({ ts: "1700001000.000100" })) - // A retry of a much older post must not un-freshen the newer engagement. - recordThreadEngagement(botPost({ ts: "1700000001.000100" })) - - const { deps, calls } = makeDeps(UNRELATED_THREAD) - // 20 minutes after the NEWER post; >30 min after the older one, so if the - // replay had won this would fall outside the recency bound. - const followUp = envelope({ event: { ts: "1700002200.000200" } }) - expect(await promoteThreadFollowUp(followUp, deps)).not.toBeNull() - expect(calls()).toBe(0) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(false) }) - test("recorded engagement is still subject to the recency bound", async () => { - recordThreadEngagement(botPost({ ts: "1700000001.000100" })) - - const { deps } = makeDeps(UNRELATED_THREAD) - // 31 minutes later: a warm cache does not make a dead thread live. - const followUp = envelope({ event: { ts: "1700001861.000200" } }) - expect(await promoteThreadFollowUp(followUp, deps)).toBeNull() - }) - - test("teams stay isolated", async () => { + test("teams stay isolated", () => { recordThreadEngagement(botPost()) - - const { deps, calls } = makeDeps(UNRELATED_THREAD) - const otherTeam = envelope({ envelope: { team_id: "T999" } }) - expect(await promoteThreadFollowUp(otherTeam, deps)).toBeNull() - expect(calls()).toBe(1) + const otherTeam = promote(envelope({ envelope: { team_id: "T999" } }))! + expect(confirmThreadFollowUp(otherTeam, UNRELATED_THREAD).engaged).toBe(false) }) - test("edits, DMs, non-events and unreadable bodies are all no-ops", async () => { + test("edits, DMs, non-events and unreadable bodies are all no-ops", () => { recordThreadEngagement(botPost({ subtype: "message_changed" })) recordThreadEngagement(botPost({ channel_type: "im" })) recordThreadEngagement(botPost({ type: "reaction_added" })) recordThreadEngagement(botPost({ ts: undefined })) recordThreadEngagement("not json") recordThreadEngagement(JSON.stringify({ type: "url_verification", challenge: "x" })) - - const { deps, calls } = makeDeps(UNRELATED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), deps)).toBeNull() - expect(calls()).toBe(1) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(false) }) - test("a file_share post (the bot uploading a chart) counts", async () => { + test("a file_share post (the bot uploading a chart) counts", () => { recordThreadEngagement(botPost({ subtype: "file_share" })) - - const { deps, calls } = makeDeps(UNRELATED_THREAD) - expect(await promoteThreadFollowUp(envelope({}), deps)).not.toBeNull() - expect(calls()).toBe(0) + expect(confirmThreadFollowUp(promote()!, UNRELATED_THREAD).engaged).toBe(true) }) }) diff --git a/apps/slack-agent/agent/lib/thread-follow-up.ts b/apps/slack-agent/agent/lib/thread-follow-up.ts index 2b6c7f803..7c69b534b 100644 --- a/apps/slack-agent/agent/lib/thread-follow-up.ts +++ b/apps/slack-agent/agent/lib/thread-follow-up.ts @@ -1,6 +1,5 @@ +import type { SlackThreadMessage } from "eve/channels/slack" import { botUserIdFromEnvelope } from "./bot-identity.js" -import { resolveBotToken } from "./maple.js" -import { emitAgentLog } from "./telemetry-log.js" import { createTtlCache } from "./ttl-cache.js" /** @@ -12,13 +11,41 @@ import { createTtlCache } from "./ttl-cache.js" * arrives as a `message.channels` / `message.groups` event and is dropped as * `unsupported` before any handler runs. Rather than patching eve's parser, * we exploit the fact that our custom `webhookVerifier` returns the body eve - * parses downstream: when a thread reply qualifies as a follow-up to a - * conversation the bot is engaged in, we rewrite `event.type` to - * `"app_mention"` so eve treats it exactly like a mention — same session - * (continuation token is `channelId:threadTs`), same incremental - * `threadContext`, same event-id dedupe. + * parses downstream: a thread reply that *could* be a follow-up gets its + * `event.type` rewritten to `"app_mention"` so eve treats it exactly like a + * mention — same session (continuation token is `channelId:threadTs`), same + * incremental `threadContext`, same event-id dedupe. * - * A reply qualifies when ALL of: + * **The promotion is optimistic, and that is the whole design.** Deciding + * whether the bot is actually engaged in a thread needs the thread, and the + * verifier is the only awaited work before eve's 200 — so anything it fetches + * is spent out of Slack's ~3s delivery budget. Two production incidents came + * out of deciding there, both surfacing as the bot going completely silent: + * + * 1. **Restart.** The engagement cache is in-memory and a deploy wipes it, so + * the first follow-up after one paid the full cold-cache round-trip + * (workspace resolve + `conversations.replies`) inside the budget, blew the + * 2s promotion deadline, and was dropped. Observed: a webhook that took + * exactly 2009ms, no turn, no log, not even the `:eyes:` ack — three + * minutes after the bot's own last post in that same thread. + * 2. **Delay.** Engagement expired after 30 minutes, and the + * `conversations.replies` fetch derived its `oldest` bound from that same + * constant. So for a reply 61.3 minutes after the bot's last post, the + * fetch window *started 31 minutes after the bot had last spoken*: the + * check could not see the bot's own messages and could only ever answer + * "not engaged". The webhook completed in 406ms — no timeout, just a + * confident wrong answer. + * + * Widening the window would have made (1) worse (a longer window is a heavier + * in-budget fetch), so both are fixed the same way: **promote first, confirm + * afterwards.** `promoteThreadFollowUp` is now synchronous and network-free — + * it parses, decides the event is a plausible candidate, registers it as + * pending, and rewrites the type. `confirmThreadFollowUp` then runs from + * `#channels/slack.js`'s handler, which eve invokes inside `waitUntil` *after* + * the 200, against the thread that handler already loads for turn context. No + * second Slack fetch, no deadline, nothing on the webhook's budget. + * + * A reply is a **candidate** when ALL of: * - it is a threaded `message` in a channel/group (not a thread root, not a * DM — DMs already dispatch on their own); * - it is user-authored (no `bot_id`, no subtype except `file_share` — @@ -26,79 +53,23 @@ import { createTtlCache } from "./ttl-cache.js" * re-triggering itself); * - it does NOT already @-mention the bot (those arrive as a separate, * real `app_mention` event; promoting the `message` twin would double- - * dispatch the same turn); - * - the bot is "engaged" in the thread: it has posted there, or someone - * mentioned it there (covers the follow-up racing ahead of the bot's - * first reply) — and that engagement is still RECENT (see - * `ENGAGEMENT_MAX_AGE_SECONDS` / `ENGAGEMENT_RECENT_MESSAGE_WINDOW`). + * dispatch the same turn). * - * Engagement is learned two ways, and the cheap one carries the load: - * `recordThreadEngagement` reads it straight off events that already prove it - * (the bot's own post echoing back, an @mention of it), while - * `isBotEngagedInThread` falls back to a `conversations.replies` round-trip for - * threads no such event has taught us about. The fallback is what has to fit - * inside Slack's webhook budget, so the fewer threads reach it, the fewer - * follow-ups are lost to `PROMOTION_DEADLINE_MS`. + * It is **confirmed** when the bot is engaged in the thread — it has posted + * there, or someone mentioned it there — and both bounds below still hold. + * Unconfirmed follow-ups are dropped by the handler throwing (eve's + * `dispatchInboundMessage` catches handler failures and abandons the turn), + * and the user is told why over DM (`#lib/disengage-notice.js`) when the bot + * had actually been working in that thread. * * Requires the Slack app to subscribe to `message.channels` (public) / - * `message.groups` (private) bot events; the engagement check reuses the - * `channels:history` / `groups:history` scopes that `threadContext` already - * needs. The bot only receives channel messages for channels it is a member - * of, which naturally bounds the event volume. + * `message.groups` (private) bot events. The bot only receives channel messages + * for channels it is a member of, which naturally bounds the event volume. */ -/** One message returned by `conversations.replies` (only what we inspect). */ -export interface ThreadReplyMessage { - readonly user?: string - readonly botId?: string - readonly text?: string - /** Slack ts ("1700000000.000100"); required for the recency bound. */ - readonly ts?: string -} - -/** Injectable dependencies so tests never touch the network. */ -export interface ThreadFollowUpDeps { - resolveBotToken(context: { teamId?: string }): Promise - fetchThreadReplies(options: { - readonly botToken: string - readonly channelId: string - readonly threadTs: string - /** - * Slack ts (epoch seconds) of the recency horizon: messages older than - * this cannot count as engagement, so the fetch itself is bounded to them - * and the trailing window is correct regardless of thread length. - */ - readonly oldest: string - /** Slack ts of the incoming reply — nothing after it can have engaged. */ - readonly latest: string - /** Aborts when the promotion deadline expires. */ - readonly signal: AbortSignal - }): Promise - /** Test-only override of `PROMOTION_DEADLINE_MS`. */ - readonly promotionDeadlineMs?: number -} - -const defaultDeps: ThreadFollowUpDeps = { - resolveBotToken, - fetchThreadReplies: fetchThreadRepliesFromSlack, -} - -// Engagement is sticky once established (the bot's reply stays in the thread -// forever), so positive entries can live long. Negative entries stay short so -// a thread the bot joins moments later is picked up quickly. What the cache -// stores is the *timestamp* of the engagement, not a boolean, so the recency -// bound below is re-applied on every hit instead of being frozen for the TTL. -// -// The TTL is measured from the last time we LEARNED of engagement, and -// `recordThreadEngagement` keeps that in step with what the bot is actually -// doing: every post of its own that echoes back through the webhook refreshes -// the entry. So a thread the bot is actively replying in never goes cold, and -// this TTL only governs threads that have genuinely gone quiet — which are -// exactly the ones worth re-verifying against Slack. -const ENGAGED_TTL_MS = 5 * 60_000 -const NOT_ENGAGED_TTL_MS = 20_000 -const CACHE_MAX_ENTRIES = 500 -const CACHE_SWEEP_INTERVAL_MS = 60_000 +// --------------------------------------------------------------------------- +// Bounds +// --------------------------------------------------------------------------- /** * Engagement expires. Without a bound, one @mention makes every later human @@ -106,31 +77,65 @@ const CACHE_SWEEP_INTERVAL_MS = 60_000 * cost amplifier and a permanently open prompt-injection intake on a thread * that has long since moved on to something else. * - * Two bounds, both must hold, both measured against the incoming reply: - * - 30 minutes since the mention / the bot's last message. A Slack thread - * that goes half an hour without the bot saying anything is a different - * conversation; re-@-mentioning it is one keystroke. - * - the engagement must be within the last 15 messages of the thread, so a - * fast-moving thread that ran away from the bot in under 30 minutes stops - * too. - * Past either bound the reply passes through unpromoted and the user simply - * @-mentions the bot again. + * Two bounds, both must hold: + * + * - **The thread is not dormant.** Measured from the message immediately + * before the reply, NOT from the bot's last post. That distinction is the + * fix for incident (2): a thread where people are still talking is a live + * conversation whether or not the bot has said anything lately, and an hour + * of human back-and-forth is a normal shape for an incident thread. What is + * genuinely over is a thread nobody has touched for a day. + * - **The bot's engagement is within the trailing 15 messages.** This is the + * real guard, and the one the clock was always a poor proxy for: a thread + * that ran away from the bot has moved on regardless of how recently it + * did so. + * + * Past either bound the reply is dropped, the humans in the thread get a DM + * saying so, and re-@-mentioning the bot is one keystroke. */ -const ENGAGEMENT_MAX_AGE_SECONDS = 30 * 60 +const THREAD_DORMANT_MAX_SECONDS = 24 * 60 * 60 const ENGAGEMENT_RECENT_MESSAGE_WINDOW = 15 /** - * Combined budget for the whole promotion side-trip on a cold cache: workspace - * resolve AND thread fetch together. Slack's webhook budget is ~3s total and - * this runs inside it, so the two calls cannot each get their own timeout - * (resolve alone is capped at 5s in maple.ts — already over budget). Past the - * deadline the event passes through unpromoted; the user can re-@-mention. + * eve's `thread.refresh()` fetches ONE oldest-first page of 50 replies, so on a + * longer thread the messages we get back are the *start* of it and the tail — + * where any recent engagement lives — is exactly what is missing. + * + * `loadThreadContextMessages` drops the triggering message from that page, so a + * full page reaches us as 49 (or as 50, when the triggering message was past + * the page and never in it). Either count means "there is more thread than we + * can see", and see `confirmThreadFollowUp` for why that promotes rather than + * drops. */ -const PROMOTION_DEADLINE_MS = 2_000 +const EVE_THREAD_PAGE_SIZE = 50 + +// --------------------------------------------------------------------------- +// Engagement cache +// --------------------------------------------------------------------------- + +// Engagement is sticky once established (the bot's reply stays in the thread +// forever), so an entry could live long; it is deliberately short instead, +// because what this cache now answers is not "has the bot ever been here" — the +// thread page answers that, for free, on the confirm path — but "is the bot +// demonstrably active here *right now*". `recordThreadEngagement` refreshes it +// from every post of the bot's own that echoes back through the webhook, so a +// thread the bot is actively replying in never goes cold, and a hit is proof +// that the bot did something here within the last ENGAGED_TTL_MS. +// +// That proof is what makes the hit a legitimate short-circuit: an entry cannot +// be older than this TTL, which is three orders of magnitude inside the +// dormancy bound, so a warm entry can never resurrect a dead thread. It also +// covers the one case the thread page structurally cannot — a thread longer +// than eve's 50-message page. +// +// There are no negative entries. They used to exist to spare a busy channel one +// `conversations.replies` call per message; the confirm path spends no network +// at all, so a cached "no" would save nothing and could only go stale. +const ENGAGED_TTL_MS = 5 * 60_000 +const CACHE_MAX_ENTRIES = 500 +const CACHE_SWEEP_INTERVAL_MS = 60_000 interface EngagementCacheEntry { - /** Slack ts (epoch seconds) of the engaging message, or null if none. */ - readonly engagedAtSeconds: number | null readonly expiresAt: number } @@ -139,43 +144,102 @@ const engagementCache = createTtlCache({ sweepIntervalMs: CACHE_SWEEP_INTERVAL_MS, }) -/** Test-only: clears the engagement cache so each test starts cold. */ -export function resetThreadEngagementCacheForTests(): void { - engagementCache.clear() -} - /** - * The cache is process-global and this app serves every workspace that + * The caches are process-global and this app serves every workspace that * installed the Slack app. Slack channel ids are only unique per workspace, so * a bare `channel:thread` key would let one tenant's engagement decide * another tenant's dispatch. */ -function engagementCacheKey( - teamId: string | undefined, - channelId: string, - threadTs: string, -): string { +function engagementCacheKey(teamId: string | undefined, channelId: string, threadTs: string): string { return `${teamId ?? "-"}:${channelId}:${threadTs}` } +// --------------------------------------------------------------------------- +// Pending-promotion registry +// --------------------------------------------------------------------------- + +/** + * A promotion the handler still has to confirm. Keyed by the message's own ts, + * which is what separates an optimistically promoted follow-up from a real + * `app_mention` once both are indistinguishable `app_mention` bodies. + * + * The alternative — re-reading `message.text` for `<@bot>` in the handler — does + * not work: eve re-renders inbound mrkdwn, so the text the handler sees is not + * necessarily the text Slack sent. + */ +export interface PendingFollowUp { + readonly teamId: string | undefined + readonly channelId: string + readonly threadTs: string + /** Slack ts of the reply itself (not the thread root). */ + readonly messageTs: string + /** The workspace's bot user id, straight off the envelope's `authorizations`. */ + readonly botUserId: string + /** + * Whether this delivery should still place the `:eyes:` ack. False for a + * Slack redelivery, where the reaction is already on the message — + * `already_reacted` is tolerated downstream, this just skips a pointless + * call. + */ + readonly ackable: boolean +} + +/** + * Long enough to cover eve's `waitUntil` dispatch (which runs immediately, + * in-process, right after the 200) with room for a queued handler, short enough + * that an event whose handler never ran does not linger. + */ +const PENDING_TTL_MS = 5 * 60_000 +const PENDING_MAX_ENTRIES = 4096 + +interface PendingEntry extends PendingFollowUp { + readonly expiresAt: number +} + +const pendingPromotions = createTtlCache({ + maxEntries: PENDING_MAX_ENTRIES, + sweepIntervalMs: CACHE_SWEEP_INTERVAL_MS, +}) + +function pendingKey(teamId: string | undefined, channelId: string, messageTs: string): string { + return `${teamId ?? "-"}:${channelId}:${messageTs}` +} + +/** Test-only: clears both caches so each test starts cold. */ +export function resetThreadFollowUpStateForTests(): void { + engagementCache.clear() + pendingPromotions.clear() +} + +/** + * The pending promotion for a dispatched message, or null when the message + * reached the handler as a real `app_mention` (or a DM) and needs no + * confirmation. + * + * Deliberately a read, not a take: a message ts identifies exactly one Slack + * message, so a re-entry for the same ts is the same follow-up and must be + * judged the same way rather than being silently upgraded to "real mention". + */ +export function pendingFollowUp(target: { + readonly teamId: string | undefined + readonly channelId: string + readonly messageTs: string +}): PendingFollowUp | null { + return pendingPromotions.get(pendingKey(target.teamId, target.channelId, target.messageTs)) ?? null +} + +// --------------------------------------------------------------------------- +// Learning engagement from the event stream +// --------------------------------------------------------------------------- + /** * Refreshes the engagement cache from an event that already *proves* the bot is * engaged — its own post echoing back through the events stream, or someone - * @-mentioning it. Free: the envelope in hand carries exactly what the - * `conversations.replies` round-trip would have gone to Slack to find out. - * - * Why this exists: engagement used to be learned only by the cold path in - * `isBotEngagedInThread`, so an entry aged out `ENGAGED_TTL_MS` after the last - * *promotion* rather than after the last thing the bot actually did. A thread - * the bot was still actively replying in could therefore hand the next - * un-mentioned follow-up a cold cache, making it pay the full two-call - * round-trip inside Slack's ~3s webhook budget — and a follow-up that misses - * `PROMOTION_DEADLINE_MS` is dropped outright, with no reply and no retry. - * That is a real incident, not a hypothetical: a follow-up was lost this way - * three minutes after the bot's own last post in the same thread. + * @-mentioning it. Free: the envelope in hand carries exactly what the thread + * page would otherwise have to be scanned for. * - * Call it on every verified inbound body, before promotion. Never throws — an - * envelope we cannot read simply teaches us nothing. + * Call it on every verified inbound body. Never throws — an envelope we cannot + * read simply teaches us nothing. */ export function recordThreadEngagement(rawBody: string): void { let parsed: unknown @@ -215,8 +279,8 @@ export function recordThreadEngagement(rawBody: string): void { const botUserId = botUserIdFromEnvelope(parsed) if (!botUserId) return - // The same predicate `latestEngagementSeconds` applies to fetched replies — - // the two must agree, or whether a thread counts as engaged would depend on + // The same predicate `lastEngagementIndex` applies to fetched thread messages + // — the two must agree, or whether a thread counts as engaged would depend on // which path happened to observe it. // // `bot_id` is deliberately NOT rejected here. `parseFollowUpCandidate` drops @@ -227,42 +291,50 @@ export function recordThreadEngagement(rawBody: string): void { const isOwnPost = typeof event.user === "string" && event.user === botUserId if (!isOwnPost && !text.includes(`<@${botUserId}>`)) return - const seconds = Number(ts) - if (!Number.isFinite(seconds)) return - // A root-level mention has no `thread_ts` yet — its own ts is what becomes // the thread's id once the bot replies, so key on that and the thread is // already warm for its very first follow-up. - const threadTs = - typeof event.thread_ts === "string" && event.thread_ts.length > 0 ? event.thread_ts : ts + const threadTs = typeof event.thread_ts === "string" && event.thread_ts.length > 0 ? event.thread_ts : ts const teamId = typeof parsed.team_id === "string" ? parsed.team_id : undefined - const key = engagementCacheKey(teamId, channelId, threadTs) - // Never move engagement backwards: Slack redelivers, and a retry carrying an - // older event must not un-freshen what a newer one has already established. - const known = engagementCache.get(key)?.engagedAtSeconds ?? null - engagementCache.set(key, { - engagedAtSeconds: known === null ? seconds : Math.max(known, seconds), + engagementCache.set(engagementCacheKey(teamId, channelId, threadTs), { expiresAt: Date.now() + ENGAGED_TTL_MS, }) } +// --------------------------------------------------------------------------- +// Promotion (webhook path — synchronous, zero network) +// --------------------------------------------------------------------------- + +export interface PromotionOptions { + /** Slack redelivery (`x-slack-retry-num` present); suppresses the `:eyes:` ack. */ + readonly isSlackRetry?: boolean +} + /** * Inspects a verified inbound Slack webhook body. Returns a rewritten body * (the same envelope with `event.type` promoted to `"app_mention"`) when the - * event is a qualifying thread follow-up, or `null` when the body should - * pass through unchanged. Never throws on malformed input — anything - * unexpected simply doesn't qualify. + * event is a plausible thread follow-up, or `null` when the body should pass + * through unchanged. + * + * Parse-only and synchronous by contract: this runs inside Slack's ~3s webhook + * budget as the last awaited step before eve's 200, and the two incidents in + * this file's header were both bought by doing more than parsing here. Never + * throws on malformed input — anything unexpected simply doesn't qualify. */ -export async function promoteThreadFollowUp( - rawBody: string, - deps: ThreadFollowUpDeps = defaultDeps, -): Promise { +export function promoteThreadFollowUp(rawBody: string, options: PromotionOptions = {}): string | null { const candidate = parseFollowUpCandidate(rawBody) if (!candidate) return null - const engaged = await isBotEngagedInThread(candidate, deps) - if (!engaged) return null + pendingPromotions.set(pendingKey(candidate.teamId, candidate.channelId, candidate.eventTs), { + teamId: candidate.teamId, + channelId: candidate.channelId, + threadTs: candidate.threadTs, + messageTs: candidate.eventTs, + botUserId: candidate.botUserId, + ackable: options.isSlackRetry !== true, + expiresAt: Date.now() + PENDING_TTL_MS, + }) candidate.envelope.event.type = "app_mention" return JSON.stringify(candidate.envelope) @@ -270,14 +342,12 @@ export async function promoteThreadFollowUp( interface FollowUpCandidate { readonly envelope: { event: { type: string } } & Record - readonly teamId?: string + readonly teamId: string | undefined readonly channelId: string readonly threadTs: string readonly botUserId: string - /** The incoming reply's own ts, verbatim — upper bound for the thread fetch. */ + /** The incoming reply's own ts, verbatim. */ readonly eventTs: string - /** The incoming reply's own ts, in epoch seconds — the recency reference. */ - readonly eventTsSeconds: number } function parseFollowUpCandidate(rawBody: string): FollowUpCandidate | null { @@ -317,15 +387,12 @@ function parseFollowUpCandidate(rawBody: string): FollowUpCandidate | null { if (!botUserId) return null // A reply that @-mentions the bot arrives as a real app_mention event too — - // promoting this twin would dispatch the same turn twice. + // promoting this twin would dispatch the same turn twice. Reading `text` is + // safe *here*, on the raw Slack envelope: it is the original mrkdwn, where a + // mention is always literally `<@Uxxxx>`. const text = typeof event.text === "string" ? event.text : "" if (text.includes(`<@${botUserId}>`)) return null - // Use the event's own clock rather than Date.now(): Slack retries deliver - // the same event minutes later, and the decision must not drift with them. - const eventTsSeconds = Number(ts) - if (!Number.isFinite(eventTsSeconds)) return null - return { envelope: parsed as FollowUpCandidate["envelope"], teamId: typeof parsed.team_id === "string" ? parsed.team_id : undefined, @@ -333,198 +400,130 @@ function parseFollowUpCandidate(rawBody: string): FollowUpCandidate | null { threadTs, botUserId, eventTs: ts, - eventTsSeconds, } } -async function isBotEngagedInThread( - candidate: FollowUpCandidate, - deps: ThreadFollowUpDeps, -): Promise { - const cacheKey = engagementCacheKey(candidate.teamId, candidate.channelId, candidate.threadTs) - const cached = engagementCache.get(cacheKey) - if (cached) return isEngagementRecent(cached.engagedAtSeconds, candidate) - - // One deadline for BOTH network round-trips (see PROMOTION_DEADLINE_MS). - const deadline = AbortSignal.timeout(deps.promotionDeadlineMs ?? PROMOTION_DEADLINE_MS) - let botToken: string - let replies: readonly ThreadReplyMessage[] - try { - // The resolve itself is deliberately NOT aborted at the deadline: it is a - // shared, de-duped, cached promise (maple.ts), so letting it finish in the - // background warms the workspace cache for the thread's next reply — we - // merely stop waiting for it here. - botToken = await withDeadline(deps.resolveBotToken({ teamId: candidate.teamId }), deadline) - replies = await withDeadline( - deps.fetchThreadReplies({ - botToken, - channelId: candidate.channelId, - threadTs: candidate.threadTs, - // Bound the fetch to the recency horizon so the trailing window is - // computed over the true tail even for threads longer than one page. - oldest: String(candidate.eventTsSeconds - ENGAGEMENT_MAX_AGE_SECONDS), - latest: candidate.eventTs, - signal: deadline, - }), - deadline, - ) - } catch (error) { - // Deadline expiry is an expected cold-cache outcome, not a failure: fall - // through unpromoted (nothing cached — the next reply retries, warmer). - // - // Expected, but never silent. This is the one path that costs a user their - // message outright: no turn is created, eve drops the event as - // `unsupported` without a line of its own, and the `:eyes:` ack does not - // fire either — so from the outside the bot simply says nothing. Left - // unlogged it is invisible to `hooks/outcome-log.ts`, which can only ever - // report on turns that exist. - if (deadline.aborted) { - emitAgentLog("warn", "follow_up_promotion_timeout", { - "maple.agent.event": "follow_up_promotion_timeout", - "maple.slack.team_id": candidate.teamId, - "maple.slack.channel_id": candidate.channelId, - "maple.slack.thread_ts": candidate.threadTs, - "maple.agent.deadline_ms": deps.promotionDeadlineMs ?? PROMOTION_DEADLINE_MS, - }) - return false - } - throw error - } +// --------------------------------------------------------------------------- +// Confirmation (dispatch path — post-200, no budget, no extra fetch) +// --------------------------------------------------------------------------- - const engagedAtSeconds = latestEngagementSeconds(replies, candidate) +export type FollowUpEngagedReason = + /** The bot posted or was mentioned here within ENGAGED_TTL_MS. */ + | "cached-engagement" + /** The thread page shows the bot inside both bounds. */ + | "recent-engagement" + /** The thread could not be read at all; promoted rather than lost. */ + | "thread-unreadable" + /** The page is a full 50 and the tail we would need is past it. */ + | "page-truncated" - engagementCache.set(cacheKey, { - engagedAtSeconds, - expiresAt: Date.now() + (engagedAtSeconds === null ? NOT_ENGAGED_TTL_MS : ENGAGED_TTL_MS), - }) - return isEngagementRecent(engagedAtSeconds, candidate) -} +/** The two bounds a follow-up can trip in a thread the bot really worked in. */ +export type FollowUpDisengagedReason = "thread-dormant" | "engagement-buried" /** - * Resolves/rejects with `promise`, but rejects as soon as `signal` aborts — - * even when the underlying work ignores the signal. The work itself is not - * cancelled; we just stop waiting for it. + * `workedInThread` is the discriminant that matters downstream: it is the + * difference between "the bot has stepped out of a conversation it was part of" + * — worth telling the humans about (`#lib/disengage-notice.js`) — and "the bot + * was never in this thread", which is just ordinary channel chatter and must + * stay silent. + */ +export type FollowUpDecision = + | { readonly engaged: true; readonly reason: FollowUpEngagedReason } + | { + readonly engaged: false + readonly workedInThread: true + readonly reason: FollowUpDisengagedReason + } + | { readonly engaged: false; readonly workedInThread: false; readonly reason: "never-engaged" } + +/** + * Decides whether an optimistically promoted follow-up should actually become a + * turn, from the thread the handler has already loaded for context. Pass `null` + * for `messages` when the thread could not be read. * - * The abort listener is deliberately NOT removed once `promise` settles, which - * looks like a leak and is not one: `{ once: true }` drops it when it fires, - * the signal is created per promotion and lives at most - * `PROMOTION_DEADLINE_MS`, and rejecting an already-settled promise is a no-op. + * Pure and synchronous: everything it needs is either in hand or in memory. * - * Removing it explicitly — the obvious tidier version — breaks this outright - * under Bun (1.3.14): taking the last `abort` listener off a signal from - * `AbortSignal.timeout()` stops its timer for good, so `.aborted` never - * becomes true and the deadline silently ceases to exist. Node is unaffected, - * so production behaved correctly and only the test runner saw it — which is - * precisely why it went unnoticed. Both calls below share one signal and the - * first one almost always settles fast, so the tidier version disarmed the - * deadline for the second, slower call: the one it exists to bound. + * **It fails open.** Where the evidence is absent rather than negative — the + * thread unreadable, or truncated past eve's page — the follow-up is promoted. + * The two outcomes are not symmetric: a wrong drop costs a user their message + * with no reply, no ack and nothing on screen to explain it (that is precisely + * how both incidents presented), while a wrong dispatch costs one turn in a + * thread the bot was already part of. */ -function withDeadline(promise: Promise, signal: AbortSignal): Promise { - if (signal.aborted) return Promise.reject(abortReason(signal)) - return new Promise((resolve, reject) => { - signal.addEventListener("abort", () => reject(abortReason(signal)), { once: true }) - promise.then(resolve, (error: unknown) => { - reject(error instanceof Error ? error : new Error(String(error))) - }) - }) -} +export function confirmThreadFollowUp( + pending: PendingFollowUp, + messages: readonly SlackThreadMessage[] | null, +): FollowUpDecision { + const cacheKey = engagementCacheKey(pending.teamId, pending.channelId, pending.threadTs) + // A warm entry is at most ENGAGED_TTL_MS old, so it is proof of activity now, + // not a remembered verdict — see the cache's own comment for why that makes + // short-circuiting safe against both bounds below. + if (engagementCache.get(cacheKey) !== undefined) { + return { engaged: true, reason: "cached-engagement" } + } -function abortReason(signal: AbortSignal): Error { - return signal.reason instanceof Error ? signal.reason : new Error("Promotion deadline expired.") -} + if (messages === null) { + return { engaged: true, reason: "thread-unreadable" } + } -/** - * Timestamp of the most recent message that engages the bot — its own post or - * an @mention of it — provided it falls inside the trailing message window. - * Returns null when the thread has none. - */ -function latestEngagementSeconds( - replies: readonly ThreadReplyMessage[], - candidate: FollowUpCandidate, -): number | null { - const mention = `<@${candidate.botUserId}>` - // `conversations.replies` is oldest-first, so the trailing window is the tail. - const recent = replies.slice(-ENGAGEMENT_RECENT_MESSAGE_WINDOW) - let latest: number | null = null - for (const message of recent) { - const engages = - message.user === candidate.botUserId || - (typeof message.text === "string" && message.text.includes(mention)) - if (!engages) continue - // A message Slack did not timestamp cannot be aged, so it cannot be - // trusted to still be recent — skip it rather than assume "now". - const seconds = Number(message.ts) - if (!Number.isFinite(seconds)) continue - if (latest === null || seconds > latest) latest = seconds + const engagementIndex = lastEngagementIndex(messages, pending.botUserId) + if (engagementIndex === -1) { + if (messages.length >= EVE_THREAD_PAGE_SIZE - 1) { + return { engaged: true, reason: "page-truncated" } + } + return { engaged: false, workedInThread: false, reason: "never-engaged" } } - return latest -} -function isEngagementRecent(engagedAtSeconds: number | null, candidate: FollowUpCandidate): boolean { - if (engagedAtSeconds === null) return false - return candidate.eventTsSeconds - engagedAtSeconds <= ENGAGEMENT_MAX_AGE_SECONDS + // Bound 1: the thread itself is still alive. Measured against the message + // immediately before this reply — `loadThreadContextMessages` returns the + // thread oldest-first and excludes the triggering message, so that is the + // last element. + const replySeconds = Number(pending.messageTs) + const previousSeconds = Number(messages[messages.length - 1]?.ts) + if ( + Number.isFinite(replySeconds) && + Number.isFinite(previousSeconds) && + replySeconds - previousSeconds > THREAD_DORMANT_MAX_SECONDS + ) { + return { engaged: false, workedInThread: true, reason: "thread-dormant" } + } + + // Bound 2: the bot is still in the thread's trailing window. The incoming + // reply counts as one of those messages — it is part of what buried the + // engagement. + const messagesSinceEngagement = messages.length - engagementIndex + 1 + if (messagesSinceEngagement > ENGAGEMENT_RECENT_MESSAGE_WINDOW) { + return { engaged: false, workedInThread: true, reason: "engagement-buried" } + } + + // Learned the expensive way (a full scan of the page, after a restart wiped + // the cache); the thread's next follow-up gets it for free. + engagementCache.set(cacheKey, { expiresAt: Date.now() + ENGAGED_TTL_MS }) + return { engaged: true, reason: "recent-engagement" } } /** - * `conversations.replies` returns oldest-first and this fetches ONE page, so - * without bounds a thread past 100 replies would have its "trailing window" - * computed over the OLDEST page: fresh engagements invisible, stale ones - * passing the recency check. `oldest`/`latest` confine the page to the recency - * horizon ending at the incoming reply, which is the only slice the engagement - * logic can act on anyway. Slack rejects JSON for this method — form-encoded - * only. + * Index of the last message that engages the bot — its own post, or an @mention + * of it — or -1 when the visible thread has none. * - * This runs INSIDE the webhook verifier, before eve returns its 200, so it - * spends Slack's ~3s delivery budget. The caller's deadline signal keeps a - * slow Slack API from turning into a retry storm: on abort the event passes - * through unpromoted. + * Identity comes from `botUserId` (learned per team from the envelope's + * `authorizations`, `#lib/bot-identity.js`), never from eve's `isMe`: that is + * `bot_id !== undefined`, i.e. ANY bot, so in a channel that also hosts a GitHub + * or CI app every one of their posts would read as ours. * - * Exported only for tests (the request-shape assertions); production reaches - * it through `defaultDeps`. + * Mentions are matched against `text`, the original Slack mrkdwn, where a + * mention is always literally `<@Uxxxx>`; `markdown` is eve's re-rendering of it + * and may have resolved the id away. */ -export async function fetchThreadRepliesFromSlack(options: { - readonly botToken: string - readonly channelId: string - readonly threadTs: string - readonly oldest: string - readonly latest: string - readonly signal: AbortSignal -}): Promise { - const res = await fetch("https://slack.com/api/conversations.replies", { - method: "POST", - headers: { - authorization: `Bearer ${options.botToken}`, - "content-type": "application/x-www-form-urlencoded", - }, - body: new URLSearchParams({ - channel: options.channelId, - ts: options.threadTs, - oldest: options.oldest, - latest: options.latest, - // `latest` is the incoming reply's own ts; inclusive so it is not - // silently dropped from the window. - inclusive: "true", - limit: "100", - }), - signal: options.signal, - }) - if (!res.ok) { - throw new Error(`Slack conversations.replies failed: HTTP ${res.status}`) - } - const payload = (await res.json()) as { - ok: boolean - error?: string - messages?: ReadonlyArray> - } - if (!payload.ok) { - throw new Error(`Slack conversations.replies failed: ${payload.error ?? "unknown_error"}`) +function lastEngagementIndex(messages: readonly SlackThreadMessage[], botUserId: string): number { + const mention = `<@${botUserId}>` + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index] + if (message === undefined) continue + if (message.user === botUserId) return index + if (message.text.includes(mention)) return index } - return (payload.messages ?? []).map((message) => ({ - user: typeof message.user === "string" ? message.user : undefined, - botId: typeof message.bot_id === "string" ? message.bot_id : undefined, - text: typeof message.text === "string" ? message.text : undefined, - ts: typeof message.ts === "string" ? message.ts : undefined, - })) + return -1 } function isRecord(value: unknown): value is Record {