Adding ContentFragment for Slack (#2003)

* [WIP] Adding ContentFragment for Slack

* Working content Fragment

* Clean up after self review

* Clean up CF names

* Clean up

* Content Fragment working, a few things missing

* Self clean up

* Adjusting content framework title

* Fix content Fragment rendering on createConversation API endpoint

* Adding ContentFragment.contentType

* Fixes

* Fix call to <Citation>

* Add channel name in content fragment title

* Bumping Sparkle to 0.2.7 in front

connectors/src/connectors/slack/bot.ts

@@ -1,3 +1,6 @@
+import { WebClient } from "@slack/web-api";
+import { Message } from "@slack/web-api/dist/response/ConversationsHistoryResponse";
+import { ConversationsRepliesResponse } from "@slack/web-api/dist/response/ConversationsRepliesResponse";
 import levenshtein from "fast-levenshtein";
 
 import {
@@ -7,6 +10,7 @@ import {
   AgentMessageType,
   ConversationType,
   DustAPI,
+  PostContentFragmentRequestBody,
   RetrievalDocumentType,
   UserMessageType,
 } from "@connectors/lib/dust_api";
@@ -21,6 +25,7 @@ import { Err, Ok, Result } from "@connectors/lib/result";
 import logger from "@connectors/logger/logger";
 
 import {
+  formatMessagesForUpsert,
   getAccessToken,
   getBotUserIdMemoized,
   getSlackClient,
@@ -137,12 +142,22 @@ async function botAnswerMessage(
     slackAvatar: null,
     channelId: slackChannel,
     messageTs: slackMessageTs,
-    threadTs: lastSlackChatBotMessage?.threadTs || slackMessageTs,
+    threadTs:
+      slackThreadTs || lastSlackChatBotMessage?.threadTs || slackMessageTs,
     conversationId: lastSlackChatBotMessage?.conversationId,
   });
 
   const accessToken = await getAccessToken(connector.connectionId);
   const slackClient = getSlackClient(accessToken);
+  // Start computing the content fragment as early as possible since it's independent from all the other I/O operations
+  // and a lot of the subsequent I/O operations can only be done sequentially.
+  const contentFragmentPromise = makeContentFragment(
+    slackClient,
+    slackChannel,
+    lastSlackChatBotMessage?.threadTs || slackThreadTs || slackMessageTs,
+    lastSlackChatBotMessage?.messageTs || slackThreadTs || slackMessageTs,
+    connector
+  );
   const slackUserInfo = await slackClient.users.info({
     user: slackUserId,
   });
@@ -309,10 +324,23 @@ async function botAnswerMessage(
     },
   };
 
+  const buildContentFragmentRes = await contentFragmentPromise;
+  if (buildContentFragmentRes.isErr()) {
+    return new Err(new Error(buildContentFragmentRes.error.message));
+  }
   let conversation: ConversationType | undefined = undefined;
   let userMessage: UserMessageType | undefined = undefined;
 
   if (lastSlackChatBotMessage?.conversationId) {
+    if (buildContentFragmentRes.value) {
+      const contentFragmentRes = await dustAPI.postContentFragment({
+        conversationId: lastSlackChatBotMessage.conversationId,
+        contentFragment: buildContentFragmentRes.value,
+      });
+      if (contentFragmentRes.isErr()) {
+        return new Err(new Error(contentFragmentRes.error.message));
+      }
+    }
     const mesasgeRes = await dustAPI.postUserMessage({
       conversationId: lastSlackChatBotMessage.conversationId,
       message: messageReqBody,
@@ -333,6 +361,7 @@ async function botAnswerMessage(
       title: null,
       visibility: "unlisted",
       message: messageReqBody,
+      contentFragment: buildContentFragmentRes.value || undefined,
     });
     if (convRes.isErr()) {
       return new Err(new Error(convRes.error.message));
@@ -523,3 +552,104 @@ export async function toggleSlackbot(
 
   return new Ok(void 0);
 }
+
+async function makeContentFragment(
+  slackClient: WebClient,
+  channelId: string,
+  threadTs: string,
+  startingAtTs: string | null,
+  connector: Connector
+) {
+  const slackChannelPromise = slackClient.conversations.info({
+    channel: channelId,
+  });
+  let allMessages: Message[] = [];
+
+  let next_cursor = undefined;
+
+  let shouldTake = false;
+  do {
+    const replies: ConversationsRepliesResponse =
+      await slackClient.conversations.replies({
+        channel: channelId,
+        ts: threadTs,
+        cursor: next_cursor,
+        limit: 100,
+      });
+    // Despite the typing, in practice `replies` can be undefined at times.
+    if (!replies) {
+      return new Err(
+        new Error(
+          "Received unexpected undefined replies from Slack API in syncThread (generally transient)"
+        )
+      );
+    }
+    if (replies.error) {
+      throw new Error(replies.error);
+    }
+    if (!replies.messages) {
+      break;
+    }
+    for (const m of replies.messages) {
+      if (m.ts === startingAtTs) {
+        // Signal that we must take all the messages starting from this one.
+        shouldTake = true;
+      }
+      if (!m.user) {
+        continue;
+      }
+      if (shouldTake) {
+        const slackChatBotMessage = await SlackChatBotMessage.findOne({
+          where: {
+            connectorId: connector.id,
+            messageTs: m.ts,
+            channelId: channelId,
+          },
+        });
+        if (slackChatBotMessage) {
+          // If this message is a mention to the bot, we don't send it as a content fragment
+          continue;
+        }
+        allMessages.push(m);
+      }
+    }
+
+    next_cursor = replies.response_metadata?.next_cursor;
+  } while (next_cursor);
+
+  const botUserId = await getBotUserIdMemoized(slackClient);
+  allMessages = allMessages.filter((m) => m.user !== botUserId);
+  if (allMessages.length === 0) {
+    return new Ok(null);
+  }
+
+  const text = await formatMessagesForUpsert(
+    channelId,
+    allMessages,
+    connector.id.toString(),
+    slackClient
+  );
+  let url: string | null = null;
+  if (allMessages[0]?.ts) {
+    const permalinkRes = await slackClient.chat.getPermalink({
+      channel: channelId,
+      message_ts: allMessages[0].ts,
+    });
+    if (!permalinkRes.ok || !permalinkRes.permalink) {
+      return new Err(new Error(permalinkRes.error));
+    }
+    url = permalinkRes.permalink;
+  }
+  const channel = await slackChannelPromise;
+  console.log(channel);
+  if (channel.error) {
+    return new Err(new Error(channel.error));
+  }
+
+  return new Ok({
+    title: `Thread content from #${channel.channel?.name}`,
+    content: text,
+    url: url,
+    contentType: "slack_thread_content",
+  } as PostContentFragmentRequestBody);
+}

connectors/src/connectors/slack/temporal/activities.ts

@@ -594,7 +594,7 @@ async function processMessageForMentions(
   return message;
 }
 
-async function formatMessagesForUpsert(
+export async function formatMessagesForUpsert(
   channelId: string,
   messages: Message[],
   connectorId: string,
@@ -649,8 +649,20 @@ export async function fetchUsers(
   } while (cursor);
 }
 
-export async function getBotUserId(slackAccessToken: string) {
-  const client = getSlackClient(slackAccessToken);
+export async function getBotUserId(
+  slackAccessToken: WebClient
+): Promise<string>;
+export async function getBotUserId(slackAccessToken: string): Promise<string>;
+export async function getBotUserId(
+  slackAccessToken: string | WebClient
+): Promise<string> {
+  let client: WebClient | undefined = undefined;
+  if (slackAccessToken instanceof WebClient) {
+    client = slackAccessToken;
+  } else {
+    client = getSlackClient(slackAccessToken);
+  }
+
   const authRes = await client.auth.test({});
   if (authRes.error) {
     throw new Error(`Failed to fetch auth info: ${authRes.error}`);

connectors/src/lib/dust_api.ts

@@ -84,6 +84,13 @@ const PostMessagesRequestBodySchemaIoTs = t.type({
   }),
 });
 
+export const PostContentFragmentRequestBodySchemaIoTs = t.type({
+  title: t.string,
+  content: t.string,
+  url: t.union([t.string, t.null]),
+  contentType: t.literal("slack_thread_content"),
+});
+
 const PostConversationsRequestBodySchemaIoTs = t.type({
   title: t.union([t.string, t.null]),
   visibility: t.union([
@@ -92,6 +99,10 @@ const PostConversationsRequestBodySchemaIoTs = t.type({
     t.literal("deleted"),
   ]),
   message: t.union([PostMessagesRequestBodySchemaIoTs, t.undefined]),
+  contentFragment: t.union([
+    PostContentFragmentRequestBodySchemaIoTs,
+    t.undefined,
+  ]),
 });
 
 type PostConversationsRequestBodySchema = t.TypeOf<
@@ -102,6 +113,10 @@ export type PostMessagesRequestBodySchema = t.TypeOf<
   typeof PostMessagesRequestBodySchemaIoTs
 >;
 
+export type PostContentFragmentRequestBody = t.TypeOf<
+  typeof PostContentFragmentRequestBodySchemaIoTs
+>;
+
 // Event sent when the user message is created.
 export type UserMessageErrorEvent = {
   type: "user_message_error";
@@ -267,6 +282,8 @@ export type TimeframeUnit = (typeof TIME_FRAME_UNITS)[number];
 
 export type AgentMessageStatus = "created" | "succeeded" | "failed";
 
+export type ContentFragmentContentType = "slack_thread_content";
+
 /**
  * Both `action` and `message` are optional (we could have a no-op agent basically).
  *
@@ -310,6 +327,19 @@ export type AgentConfigurationType = {
   name: string;
 };
 
+export type ContentFragmentType = {
+  id: ModelId;
+  sId: string;
+  created: number;
+  type: "content_fragment";
+  visibility: MessageVisibility;
+  version: number;
+
+  title: string;
+  content: string;
+  contentType: ContentFragmentContentType;
+};
+
 export class DustAPI {
   _credentials: DustAPICredentials;
 
@@ -354,6 +384,7 @@ export class DustAPI {
     title,
     visibility,
     message,
+    contentFragment,
   }: PostConversationsRequestBodySchema): Promise<
     Result<
       { conversation: ConversationType; message: UserMessageType },
@@ -372,6 +403,7 @@ export class DustAPI {
           title,
           visibility,
           message,
+          contentFragment,
         }),
       }
     );
@@ -564,4 +596,33 @@ export class DustAPI {
     }
     return new Ok(json.agentConfigurations as AgentConfigurationType[]);
   }
+
+  async postContentFragment({
+    conversationId,
+    contentFragment,
+  }: {
+    conversationId: string;
+    contentFragment: PostContentFragmentRequestBody;
+  }) {
+    const res = await fetch(
+      `${DUST_API}/api/v1/w/${this.workspaceId()}/assistant/conversations/${conversationId}/content_fragments`,
+      {
+        method: "POST",
+        headers: {
+          Authorization: `Bearer ${this._credentials.apiKey}`,
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({
+          ...contentFragment,
+        }),
+      }
+    );
+
+    const json = await res.json();
+
+    if (json.error) {
+      return new Err(json.error as DustAPIErrorResponse);
+    }
+    return new Ok(json.contentFragment as ContentFragmentType);
+  }
 }

front/components/assistant/conversation/AgentMessage.tsx

@@ -271,6 +271,7 @@ export function AgentMessage({
       buttons={buttons}
       avatarBusy={agentMessageToRender.status === "created"}
       reactions={reactions}
+      enableEmojis={true}
     >
       {renderMessage(agentMessageToRender, references, shouldStream)}
     </ConversationMessage>

front/components/assistant/conversation/ContentFragment.tsx

@@ -0,0 +1,23 @@
+import { Citation } from "@dust-tt/sparkle";
+
+import { assertNever } from "@app/lib/utils";
+import { ContentFragmentType } from "@app/types/assistant/conversation";
+
+export function ContentFragment({ message }: { message: ContentFragmentType }) {
+  let logoType: "document" | "slack" = "document";
+  switch (message.contentType) {
+    case "slack_thread_content":
+      logoType = "slack";
+      break;
+
+    default:
+      assertNever(message.contentType);
+  }
+  return (
+    <Citation
+      title={message.title}
+      type={logoType}
+      href={message.url || undefined}
+    />
+  );
+}

front/components/assistant/conversation/Conversation.tsx

@@ -21,6 +21,8 @@ import {
 } from "@app/types/assistant/conversation";
 import { UserType, WorkspaceType } from "@app/types/user";
 
+import { ContentFragment } from "./ContentFragment";
+
 export default function Conversation({
   owner,
   user,
@@ -204,7 +206,16 @@ export default function Conversation({
               </div>
             );
           case "content_fragment":
-            return null;
+            return (
+              <div
+                key={`message-id-${m.sId}`}
+                className="items-center border-t bg-structure-50 px-2 py-6"
+              >
+                <div className="mx-auto flex max-w-4xl flex-col gap-4">
+                  <ContentFragment message={m} />
+                </div>
+              </div>
+            );
           default:
             ((message: never) => {
               console.error("Unknown message type", message);