diff --git a/doc/connections/SLACK-AUTOMATIC-SETUP.md b/doc/connections/SLACK-AUTOMATIC-SETUP.md index 8ac45b0b23..f15ee5f91b 100644 --- a/doc/connections/SLACK-AUTOMATIC-SETUP.md +++ b/doc/connections/SLACK-AUTOMATIC-SETUP.md @@ -102,6 +102,16 @@ subscribes to the Slack thread, and starts the agent; later replies in that thre continue the same task without another mention. Normal account and channel access rules still apply, and repeated Slack deliveries do not create duplicate tasks. +A native Slack thread keeps the same task in channels and DMs, including after +the task is Done or Cancelled. A new reply reopens that task with its saved +comments. DM access settings and task controls still apply to threads inside DMs. +An explicit `/new` or `/close` starts the next task generation only after its +control response is delivered. Task completion alone does not release the binding. +Top-level DMs keep their linear lifecycle and can start a new task after completion. +For a thread that was split before this fix, later replies continue its latest +saved task. Earlier tasks remain separate; this change does not merge history +or copy messages from the surrounding DM into a new thread. + Agents can ask interactive questions in Slack through Paperclip's saved `ask_user_questions` interactions. Native runners use `request_human_input`; legacy adapters use the issue-interactions API documented in the injected diff --git a/maps/feature-map/chat-channels.md b/maps/feature-map/chat-channels.md index 977f0fb030..679eb9265e 100644 --- a/maps/feature-map/chat-channels.md +++ b/maps/feature-map/chat-channels.md @@ -101,9 +101,9 @@ cover the database and prompt contracts; they do not prove live provider deliver For Slack, enable **Require at-mention** and **Allow direct messages**. Send a plain DM and follow up without a mention: both should reach the agent, with -top-level replies to top-level DMs. Channel messages and thread replies still +top-level replies to top-level DMs. Channel messages and channel thread replies still require a mention when that setting is enabled. Disable direct messages and -confirm a plain DM is rejected. Use a private channel for threaded conversations. +confirm a plain DM is rejected. Threads also work in DMs. For Slack acknowledgement latency, mention the bot with and without an attachment. Eyes should appear after authorized task/comment persistence, independently of diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 0d8720e58a..b35cdbda55 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -30541,6 +30541,82 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await service.shutdown(); }); + it.each([ + { isDM: true, threaded: true, status: "done", state: "active" }, + { isDM: true, threaded: true, status: "cancelled", state: "active" }, + { isDM: true, threaded: true, status: "in_review", state: "completed" }, + { isDM: false, threaded: true, status: "done", state: "completed" }, + { isDM: true, threaded: false, status: "done", state: "completed" }, + ] as const)( + "preserves Slack thread history across completion ($isDM, $threaded, $status, $state)", + async ({ isDM, threaded, status, state }) => { + const fixture = await seedCompany(); + const { callbacks, endpoint, service, wakeup } = await configuredSlackEndpoint(fixture, { linkedBoardUser: true }); + const channelId = isDM ? "D-HISTORY" : "C-HISTORY"; + const thread = makeThread({ + channelId, + id: `slack:${channelId}:${threaded ? "1000.1" : ""}`, + isDM, + }).thread; + const send = (id: string, text: string) => deliverMessage({ + callbacks, endpointId: endpoint.id, thread, + message: makeMessage({ id, text, mentioned: !isDM, userId: "UBOARD" }), + trigger: isDM ? "direct_message" : "mention", + }); + await send("1000.1", "Keep this context for my next request"); + const [original] = await readChatConversationsByEndpointId(endpoint.id); + await db.insert(issueComments).values({ + companyId: fixture.companyId, issueId: original.issueId, + authorType: "agent", authorAgentId: fixture.assignedAgentId, + body: "I kept the original request.", + }); + await db.update(issues).set({ status }).where(eq(issues.id, original.issueId)); + await db.update(chatConversations).set({ state }).where(eq(chatConversations.id, original.id)); + + await send("1000.2", "Is it done?"); + await send("1000.2", "Is it done?"); + const conversations = await readChatConversationsByEndpointId(endpoint.id); + expect(conversations).toHaveLength(threaded ? 1 : 2); + const generation = threaded ? original.sessionGeneration : original.sessionGeneration + 1; + const continued = conversations.find(row => row.sessionGeneration === generation)!; + expect(continued.issueId === original.issueId).toBe(threaded); + expect(continued.state).toBe("active"); + expect((await issueService(db).getById(continued.issueId))?.status).toBe(status === "in_review" ? "in_review" : "todo"); + const comments = await readIssueCommentsByIssueId(continued.issueId) + .orderBy(asc(issueComments.createdAt), asc(issueComments.id)); + expect(comments.map(row => row.body)).toEqual(threaded + ? ["Keep this context for my next request", "I kept the original request.", "Is it done?"] + : ["Is it done?"]); + expect(wakeup).toHaveBeenCalledTimes(2); + const resources = await db.select().from(chatEndpointResources) + .where(eq(chatEndpointResources.endpointId, endpoint.id)); + expect(resources.find(row => row.providerResourceId === channelId)?.type).toBe(isDM ? "direct_message" : "channel"); + if (isDM) { + await service.update(endpoint.id, { allowDirectMessages: false }); + await send("1000.3", "This must be rejected"); + expect(wakeup).toHaveBeenCalledTimes(2); + expect(await readIssueCommentsByIssueId(continued.issueId)).toHaveLength(threaded ? 3 : 1); + } + }, + ); + + it("keeps separate Slack DM thread roots on separate tasks", async () => { + const fixture = await seedCompany(); + const { callbacks, endpoint } = await configuredSlackEndpoint(fixture); + for (const root of ["1000.1", "1000.2"]) { + await deliverMessage({ + callbacks, endpointId: endpoint.id, + thread: makeThread({ channelId: "D-ROOTS", id: `slack:D-ROOTS:${root}`, isDM: true }).thread, + message: makeMessage({ id: root, text: "A separate request" }), + trigger: "direct_message", + }); + } + const conversations = await readChatConversationsByEndpointId(endpoint.id); + expect(conversations).toHaveLength(2); + expect(new Set(conversations.map(row => row.issueId)).size).toBe(2); + expect(conversations.map(row => row.sessionGeneration)).toEqual([1, 1]); + }); + it("deduplicates inbound events, keeps one task per thread, and enables newly discovered Slack channels", async () => { const fixture = await seedCompany(); const { callbacks, endpoint, runtime, service, wakeup } = @@ -36300,10 +36376,11 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await resumed.service.shutdown(); }); - it("returns two answered Slack turns to Idle in the same thread", async () => { + it.each([false, true])("returns two answered Slack turns to Idle in the same thread (DM=%s)", async (isDM) => { const fixture = await seedCompany(); const { callbacks, endpoint, service } = await configuredSlackEndpoint(fixture); - const thread = makeThread({ channelId: "C-IDLE", id: "slack:C-IDLE:4049.1", name: "idle" }); + const channelId = isDM ? "D-IDLE" : "C-IDLE"; + const thread = makeThread({ channelId, id: `slack:${channelId}:4049.1`, name: "idle", isDM }); let issueId: string | undefined; let conversationId: string | undefined; for (const [index, text] of ["@maya are you there?", "what is your name?"].entries()) { @@ -58532,7 +58609,14 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ]); }); - it("binds Slack DM slash controls to the native root returned for the latest task", async () => { + it.each([ + { control: "new", status: "todo", state: "active" }, + { control: "close", status: "todo", state: "active" }, + { control: "new", status: "done", state: "active" }, + { control: "close", status: "cancelled", state: "active" }, + { control: "new", status: "done", state: "completed" }, + { control: "close", status: "cancelled", state: "completed" }, + ] as const)("binds Slack DM slash controls to the latest task ($control, $status, $state)", async ({ control, status, state }) => { const fixture = await seedCompany(); const { callbacks, endpoint, runtime, service } = await configuredSlackEndpoint(fixture); @@ -58589,18 +58673,20 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ); const [conversation] = await service.listConversations(endpoint.id); expect(conversation.externalThreadId).toBe(starterThreadId); + await db.update(issues).set({ status }).where(eq(issues.id, conversation.issueId)); + await db.update(chatConversations).set({ state }).where(eq(chatConversations.id, conversation.id)); await invoke("status"); await service.processPendingPublications(); expect(runtime.endpoints.get(endpoint.id)?.posts.at(-1)).toMatchObject({ threadId: starterThreadId, - text: expect.stringMatching(/— todo$/), + text: expect.stringMatching(new RegExp(`— ${status}$`)), }); - await invoke("close"); - await expect(service.listConversations(endpoint.id)).resolves.toEqual([ + await invoke(control); + await expect(readChatConversationsByEndpointId(endpoint.id)).resolves.toEqual([ expect.objectContaining({ externalThreadId: starterThreadId, - state: "active", + state, }), ]); await service.processPendingPublications(); @@ -58610,9 +58696,26 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { state: "completed", }), ]); - expect( - await readIssuesByCompanyId(fixture.companyId), - ).toHaveLength(1); + const [completion] = await db.select().from(chatPublications).where(and( + eq(chatPublications.conversationId, conversation.id), + like(chatPublications.idempotencyKey, `control:${control}:%`), + )); + expect(completion?.state).toBe("published"); + expect(await readIssuesByCompanyId(fixture.companyId)).toHaveLength(1); + const providerTime = Date.now(); + const providerTimestamp = `${Math.floor(providerTime / 1_000)}.${String((providerTime % 1_000) * 1_000).padStart(6, "0")}`; + await deliverMessage({ + callbacks, endpointId: endpoint.id, + thread: makeThread({ channelId: "D09SLASHTASK", id: starterThreadId, isDM: true }).thread, + message: makeMessage({ id: providerTimestamp, raw: { ts: providerTimestamp }, text: "An explicitly requested new task", userId: "U-DM-SLASH-TASK" }), + trigger: "direct_message", + }); + const conversations = await service.listConversations(endpoint.id); + expect(conversations).toHaveLength(2); + expect(conversations.find(row => row.state === "active")).toMatchObject({ + externalThreadId: starterThreadId, sessionGeneration: 2, + }); + expect(await readIssuesByCompanyId(fixture.companyId)).toHaveLength(2); }); it("acknowledges a Slack retry while one durable worker owns the unrecorded command root", async () => { diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index 00a3a307e8..bbc5a8b18b 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1755,13 +1755,17 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { let { conversation: existingConversation, issue: existingIssue, latestConversation, isLinear } = await resolveInboundConversationBinding(db, { endpoint, thread, - isLinear: () => endpoint.provider === "speko" ? SPEKO_CONVERSATION_IS_LINEAR : surfaceKind !== "native_thread", + isLinear: () => endpoint.provider === "speko" + ? SPEKO_CONVERSATION_IS_LINEAR + : surfaceKind !== "native_thread" || (endpoint.provider === "slack" && thread.isDM), lookupConversation: endpoint.provider === "microsoft-teams" ? () => conversationForThread(endpoint.id, thread.id) : undefined, conversationIsComplete: endpoint.provider === "imessage-photon" ? (conversationId) => photonConversationBindingIsComplete(conversationId, hasCommittedTaskControlCompletion) - : undefined, + : endpoint.provider === "slack" && thread.isDM && surfaceKind === "native_thread" + ? async (conversationId, state) => state === "completed" && hasCommittedTaskControlCompletion(conversationId) + : undefined, }); const telegramControls = endpoint.provider === "telegram" ? telegramInboundControls(message.text, surfaceKind) diff --git a/server/src/services/chat-channels/inbound/authorization.ts b/server/src/services/chat-channels/inbound/authorization.ts index cb3f330a3d..2817378aeb 100644 --- a/server/src/services/chat-channels/inbound/authorization.ts +++ b/server/src/services/chat-channels/inbound/authorization.ts @@ -373,6 +373,7 @@ export function createChatInboundAuthority(db: Db, options: InboundAuthorityOpti const type = providerResourceType( endpoint.provider, chatSurfaceKind(endpoint.provider, thread), + thread.isDM, ); const providerResourceId = canonicalProviderResourceId( endpoint.provider, diff --git a/server/src/services/chat-channels/inbound/conversation-binding.ts b/server/src/services/chat-channels/inbound/conversation-binding.ts index f177aabd8e..c79f87aae6 100644 --- a/server/src/services/chat-channels/inbound/conversation-binding.ts +++ b/server/src/services/chat-channels/inbound/conversation-binding.ts @@ -9,7 +9,7 @@ type ConversationBindingInput = { thread: { id: string; channelId: string }; isLinear: () => boolean; lookupConversation?: () => Promise; - conversationIsComplete?: (conversationId: string) => Promise; + conversationIsComplete?: (conversationId: string, state: ConversationRow["state"]) => Promise; }; export async function resolveInboundConversationBinding( @@ -54,7 +54,7 @@ export async function resolveInboundConversationBinding( isLinear && existingConversation && (conversationIsComplete - ? await conversationIsComplete(existingConversation.id) + ? await conversationIsComplete(existingConversation.id, existingConversation.state) : existingConversation.state === "completed" || existingIssue?.status === "done" || existingIssue?.status === "cancelled") diff --git a/server/src/services/chat-channels/inbound/envelope.ts b/server/src/services/chat-channels/inbound/envelope.ts index 8bc459e5c7..4fd2fc32a6 100644 --- a/server/src/services/chat-channels/inbound/envelope.ts +++ b/server/src/services/chat-channels/inbound/envelope.ts @@ -263,7 +263,7 @@ export function createInboundEnvelope(db: Db, options: InboundEnvelopeOptions) { handle: message.author.userName, }, resource: { - type: providerResourceType(endpoint.provider, surfaceKind), + type: providerResourceType(endpoint.provider, surfaceKind, thread.isDM), providerResourceId: canonicalProviderResourceId( endpoint.provider, thread, @@ -340,7 +340,7 @@ export function createInboundEnvelope(db: Db, options: InboundEnvelopeOptions) { } : {}), resource: { - type: providerResourceType(endpoint.provider, surfaceKind), + type: providerResourceType(endpoint.provider, surfaceKind, thread.isDM), providerResourceId: canonicalProviderResourceId( endpoint.provider, thread, diff --git a/server/src/services/chat-channels/inbound/surfaces.ts b/server/src/services/chat-channels/inbound/surfaces.ts index 7f89d794c6..22e8392004 100644 --- a/server/src/services/chat-channels/inbound/surfaces.ts +++ b/server/src/services/chat-channels/inbound/surfaces.ts @@ -1,4 +1,4 @@ -import { slackThreadChannelId } from "../slack/messages.js"; +import { slackThreadChannelId, slackThreadRootMessageId } from "../slack/messages.js"; import type { EndpointRow } from "../runtime-types.js"; @@ -30,8 +30,9 @@ type ChatSurfaceKind = "direct_message" | "linear_group" | "native_thread"; function providerResourceType( provider: ChatProvider, surfaceKind: ChatSurfaceKind, + isDirectMessage: boolean, ): string { - if (surfaceKind === "direct_message") return "direct_message"; + if (isDirectMessage) return "direct_message"; if (provider === "github") return "repository"; if (provider === "imessage-photon") return "group_chat"; if (provider === "discord") return "channel"; @@ -116,6 +117,7 @@ function chatSurfaceKind( provider: ChatProvider, thread: Thread, ): ChatSurfaceKind { + if (provider === "slack" && slackThreadRootMessageId(thread.id)) return "native_thread"; if (thread.isDM) return "direct_message"; if (provider === "imessage-photon") return "linear_group"; if (provider === "telegram") { diff --git a/server/src/services/chat-channels/slack/messages.ts b/server/src/services/chat-channels/slack/messages.ts index e492cf1863..36196eab18 100644 --- a/server/src/services/chat-channels/slack/messages.ts +++ b/server/src/services/chat-channels/slack/messages.ts @@ -91,8 +91,12 @@ export function slackThreadChannelId(threadId: string): string | null { return channelId || null; } +export function slackThreadRootMessageId(threadId: string): string | null { + return /^slack:[^:]+:([^:]+)$/.exec(threadId)?.[1] ?? null; +} + export function slackOrphanFollowUp(threadId: string, messageId: string): boolean { - const rootMessageId = /^slack:[^:]+:(.+)$/.exec(threadId)?.[1]; + const rootMessageId = slackThreadRootMessageId(threadId); return Boolean(rootMessageId) && rootMessageId !== messageId; } diff --git a/server/src/services/chat-channels/slack/task-start.ts b/server/src/services/chat-channels/slack/task-start.ts index 309953a69f..8fd1251d75 100644 --- a/server/src/services/chat-channels/slack/task-start.ts +++ b/server/src/services/chat-channels/slack/task-start.ts @@ -19,7 +19,6 @@ import { inArray, lte, ne, - notInArray, or, sql, } from "drizzle-orm"; @@ -173,8 +172,6 @@ export function createSlackTaskStart(db: Db, options: SlackTaskStartOptions) { eq(chatConversations.endpointId, endpointId), inArray(chatConversations.externalConversationId, conversationIds), eq(chatConversations.isDirectMessage, true), - inArray(chatConversations.state, ["active", "waiting"]), - notInArray(issues.status, ["done", "cancelled"]), ), ) .orderBy(