fix(slack): retain DM thread tasks after completion

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/__tests__/chat-channels.integration.test.ts

@@ -30531,6 +30531,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 } =
@@ -36290,10 +36366,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()) {

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,

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,

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") {

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;
 }