DATAGO-88877: (Chore) Enable Whitesource Scan on Merge to Main

.github/workflows/ci.yaml

@@ -14,7 +14,13 @@ permissions:
 
 jobs:
   ci:
-    uses: SolaceDev/solace-public-workflows/.github/workflows/[email protected]
+    uses: SolaceDev/solace-public-workflows/.github/workflows/hatch_ci.yml@main
+    with:
+      min-python-version: "3.10"
+      whitesource_product_name: "ai"
     secrets:
       SONAR_TOKEN: ${{ secrets.SONAR_TOKEN }}
       SONAR_HOST_URL: ${{ vars.SONAR_HOST_URL }}
+      WHITESOURCE_API_KEY: ${{ secrets.WHITESOURCE_API_KEY }}
+      MANIFEST_AWS_ACCESS_KEY_ID: ${{ secrets.MANIFEST_READ_ONLY_AWS_ACCESS_KEY_ID }}
+      MANIFEST_AWS_SECRET_ACCESS_KEY: ${{ secrets.MANIFEST_READ_ONLY_AWS_SECRET_ACCESS_KEY }}

.github/workflows/release.yaml

@@ -18,9 +18,11 @@ permissions:
 
 jobs:
   release:
-    uses: SolaceDev/solace-public-workflows/.github/workflows/hatch_release_pypi.yml@v1.0.0
+    uses: SolaceDev/solace-public-workflows/.github/workflows/hatch_release_pypi.yml@main
     with:
+      ENVIRONMENT: pypi
       version: ${{ github.event.inputs.version }}
       pypi-project: solace-ai-connector-slack
     secrets:
       COMMIT_KEY: ${{ secrets.COMMIT_KEY }}
+      PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}

pyproject.toml

@@ -11,7 +11,7 @@ authors = [
 ]
 description = "Slack plugin for the Solace AI Connector - this provides an input and output component to talk to Slack"
 readme = "README.md"
-requires-python = ">=3.8"
+requires-python = ">=3.10"
 classifiers = [
     "Programming Language :: Python :: 3",
     "License :: OSI Approved :: MIT License",
@@ -20,14 +20,20 @@ classifiers = [
 dependencies = [
     "PyYAML>=6.0.1",
     "slack_bolt>=1.18.1",
-    "solace_ai_connector>=0.1.1",
+    "solace_ai_connector>=0.1.3",
+    "prettytable>=3.10.0",
 ]
 
 [project.urls]
 homepage = "https://github.com/SolaceLabs/solace-ai-connector-slack"
 repository = "https://github.com/SolaceLabs/solace-ai-connector-slack"
 documentation = "https://github.com/SolaceLabs/solace-ai-connector-slack/blob/main/docs/index.md"
 
+[tool.hatch.envs.hatch-test]
+installer = "pip"
+
+[[tool.hatch.envs.hatch-test.matrix]]
+python = ["3.10", "3.12"]
 
 [tool.hatch.build.targets.wheel]
 packages = ["src/solace_ai_connector_slack"]

src/solace_ai_connector_slack/components/slack_input.py

@@ -267,8 +267,26 @@ def handle_event(self, event):
         except Exception as e:
             log.error("Error getting team domain: %s", e)
 
-        user_email = self.get_user_email(event["user"])
+        user_email = self.get_user_email(event.get("user"))
         (text, mention_emails) = self.process_text_for_mentions(event["text"])
+
+        # Determine the reply_to thread to put in the message
+        if (
+            event.get("channel_type") == "im"
+            and event.get("subtype", event.get("type")) == "app_mention"
+        ):
+            # First message uses ts, subsequent messages use thread_ts
+            reply_to = event.get("thread_ts") or event.get("ts")
+        else:
+            # First message uses ts, subsequent messages use thread_ts
+            # thread_ts is null in direct messages
+            reply_to = event.get("thread_ts")
+
+        if reply_to:
+            thread_id = f"{event.get('channel')}_{reply_to}"
+        else:
+            thread_id = event.get("channel")
+
         payload = {
             "text": text,
             "files": files,
@@ -278,32 +296,50 @@ def handle_event(self, event):
             "mentions": mention_emails,
             "type": event.get("type"),
             "client_msg_id": event.get("client_msg_id"),
-            "ts": event.get("thread_ts"),
+            "ts": event.get("ts"),
             "channel": event.get("channel"),
             "channel_name": event.get("channel_name", ""),
             "subtype": event.get("subtype"),
             "event_ts": event.get("event_ts"),
+            "thread_ts": event.get("thread_ts"),
             "channel_type": event.get("channel_type"),
             "user_id": event.get("user"),
+            "thread_id": thread_id,
+            "reply_to_thread": reply_to,
         }
         user_properties = {
             "user_email": user_email,
             "team_id": event.get("team"),
             "type": event.get("type"),
             "client_msg_id": event.get("client_msg_id"),
-            "ts": event.get("thread_ts"),
+            "ts": event.get("ts"),
+            "thread_ts": event.get("thread_ts"),
             "channel": event.get("channel"),
             "subtype": event.get("subtype"),
             "event_ts": event.get("event_ts"),
             "channel_type": event.get("channel_type"),
             "user_id": event.get("user"),
+            "input_type": "slack",
+            "thread_id": thread_id,
+            "reply_to_thread": reply_to,
         }
 
-        if self.acknowledgement_message:
+        if self.acknowledgement_message and event.get("channel_type") == "im":
             ack_msg_ts = self.app.client.chat_postMessage(
                 channel=event["channel"],
                 text=self.acknowledgement_message,
-                thread_ts=event.get("thread_ts"),
+                blocks=[
+                    {
+                        "type": "context",
+                        "elements": [
+                            {
+                                "type": "mrkdwn",
+                                "text": self.acknowledgement_message,
+                            }
+                        ],
+                    }
+                ],
+                thread_ts=reply_to,
             ).get("ts")
             user_properties["ack_msg_ts"] = ack_msg_ts
 
@@ -323,6 +359,8 @@ def get_user_email(self, user_id):
 
     def process_text_for_mentions(self, text):
         mention_emails = []
+        if "<@" not in text:
+            return text, mention_emails
         for mention in text.split("<@"):
             if mention.startswith("!"):
                 mention = mention[1:]
@@ -420,7 +458,6 @@ def handle_new_channel_join(self, event):
     def register_handlers(self):
         @self.app.event("message")
         def handle_chat_message(event):
-            print("Got message event: ", event, event.get("channel_type"))
             if event.get("channel_type") == "im":
                 self.handle_event(event)
             elif event.get("channel_type") == "channel":

src/solace_ai_connector_slack/components/slack_output.py

@@ -1,6 +1,8 @@
 import base64
 import re
+from datetime import datetime
 
+from prettytable import PrettyTable
 
 from solace_ai_connector.common.log import log
 from .slack_base import SlackBase
@@ -114,77 +116,150 @@ class SlackOutput(SlackBase):
     def __init__(self, **kwargs):
         super().__init__(info, **kwargs)
         self.fix_formatting = self.get_config("correct_markdown_formatting", True)
+        self.streaming_state = {}
 
     def invoke(self, message, data):
-        message_info = data.get("message_info")
         content = data.get("content")
+        message_info = data.get("message_info")
+
         text = content.get("text")
-        stream = content.get("stream")
-        channel = message_info.get("channel")
+        uuid = content.get("uuid")
+        files = content.get("files")
+        streaming = content.get("streaming")
+        status_update = content.get("status_update")
+        response_complete = content.get("response_complete")
+        last_chunk = content.get("last_chunk")
+        first_chunk = content.get("first_chunk")
+
         thread_ts = message_info.get("ts")
+        channel = message_info.get("channel")
         ack_msg_ts = message_info.get("ack_msg_ts")
 
+        if response_complete:
+            status_update = True
+            text = ":checkered_flag: Response complete"
+        elif status_update:
+            text = ":thinking_face: " + text
+
+        if not channel:
+            log.error("slack_output: No channel specified in message")
+            self.discard_current_message()
+            return None
+
         return {
-            "channel": channel,
             "text": text,
-            "files": content.get("files"),
+            "uuid": uuid,
+            "files": files,
+            "streaming": streaming,
+            "channel": channel,
             "thread_ts": thread_ts,
             "ack_msg_ts": ack_msg_ts,
-            "stream": stream,
+            "status_update": status_update,
+            "last_chunk": last_chunk,
+            "first_chunk": first_chunk,
         }
 
     def send_message(self, message):
         try:
             channel = message.get_data("previous:channel")
             messages = message.get_data("previous:text")
-            stream = message.get_data("previous:stream")
+            streaming = message.get_data("previous:streaming")
             files = message.get_data("previous:files") or []
-            thread_ts = message.get_data("previous:ts")
+            reply_to = (message.get_user_properties() or {}).get("reply_to_thread", message.get_data("previous:thread_ts"))
             ack_msg_ts = message.get_data("previous:ack_msg_ts")
+            first_chunk = message.get_data("previous:first_chunk")
+            last_chunk = message.get_data("previous:last_chunk")
+            uuid = message.get_data("previous:uuid")
+            status_update = message.get_data("previous:status_update")
 
             if not isinstance(messages, list):
                 if messages is not None:
                     messages = [messages]
                 else:
                     messages = []
 
-            for text in messages:
+            for index, text in enumerate(messages):
+                if not text or not isinstance(text, str):
+                    continue
+
                 if self.fix_formatting:
                     text = self.fix_markdown(text)
-                if stream:
-                    if ack_msg_ts:
+
+                if index != 0:
+                    text = "\n" + text
+
+                if first_chunk:
+                    streaming_state = self.add_streaming_state(uuid)
+                else:
+                    streaming_state = self.get_streaming_state(uuid)
+                    if not streaming_state:
+                        streaming_state = self.add_streaming_state(uuid)
+
+                if streaming:
+                    if streaming_state.get("completed"):
+                        # We can sometimes get a message after the stream has completed
+                        continue
+
+                    streaming_state["completed"] = last_chunk
+                    ts = streaming_state.get("ts")
+                    if status_update:
+                        blocks = [
+                            {
+                                "type": "context",
+                                "elements": [
+                                    {
+                                        "type": "mrkdwn",
+                                        "text": text,
+                                    }
+                                ],
+                            },
+                        ]
+
+                        if not ts:
+                            ts = ack_msg_ts
                         try:
                             self.app.client.chat_update(
-                                channel=channel, ts=ack_msg_ts, text=text
+                                channel=channel, ts=ts, text="test", blocks=blocks
+                            )
+                        except Exception:
+                            pass
+                    elif ts:
+                        try:
+                            self.app.client.chat_update(
+                                channel=channel, ts=ts, text=text
                             )
                         except Exception:
                             # It is normal to possibly get an update after the final
                             # message has already arrived and deleted the ack message
                             pass
+                    else:
+                        response = self.app.client.chat_postMessage(
+                            channel=channel, text=text, thread_ts=reply_to
+                        )
+                        streaming_state["ts"] = response["ts"]
+
                 else:
-                    self.app.client.chat_postMessage(
-                        channel=channel, text=text, thread_ts=thread_ts
-                    )
+                    # Not streaming
+                    ts = streaming_state.get("ts")
+                    streaming_state["completed"] = True
+                    if not ts:
+                        self.app.client.chat_postMessage(
+                            channel=channel, text=text, thread_ts=reply_to
+                        )
 
             for file in files:
                 file_content = base64.b64decode(file["content"])
                 self.app.client.files_upload_v2(
                     channel=channel,
                     file=file_content,
-                    thread_ts=thread_ts,
+                    thread_ts=reply_to,
                     filename=file["name"],
                 )
         except Exception as e:
             log.error("Error sending slack message: %s", e)
 
         super().send_message(message)
 
-        try:
-            if ack_msg_ts and not stream:
-                self.app.client.chat_delete(channel=channel, ts=ack_msg_ts)
-        except Exception:
-            pass
-
     def fix_markdown(self, message):
         # Fix links - the LLM is very stubborn about giving markdown links
         # Find [text](http...) and replace with <http...|text>
@@ -193,4 +268,55 @@ def fix_markdown(self, message):
         message = re.sub(r"```[a-z]+\n", "```", message)
         # Fix bold
         message = re.sub(r"\*\*(.*?)\*\*", r"*\1*", message)
+
+        # Reformat a table to be Slack compatible
+        message = self.convert_markdown_tables(message)
+
         return message
+
+    def get_streaming_state(self, uuid):
+        return self.streaming_state.get(uuid)
+
+    def add_streaming_state(self, uuid):
+        state = {
+            "create_time": datetime.now(),
+        }
+        self.streaming_state[uuid] = state
+        self.age_out_streaming_state()
+        return state
+
+    def delete_streaming_state(self, uuid):
+        try:
+            del self.streaming_state[uuid]
+        except KeyError:
+            pass
+
+    def age_out_streaming_state(self, age=60):
+        # Note that we can later optimize this by using an array of streaming_state that
+        # is ordered by create_time and then we can just remove the first element until
+        # we find one that is not expired.
+        now = datetime.now()
+        for uuid, state in list(self.streaming_state.items()):
+            if (now - state["create_time"]).total_seconds() > age:
+                del self.streaming_state[uuid]
+
+    def convert_markdown_tables(self, message):
+        def markdown_to_fixed_width(match):
+            table_str = match.group(0)
+            rows = [
+                line.strip().split("|")
+                for line in table_str.split("\n")
+                if line.strip()
+            ]
+            headers = [cell.strip() for cell in rows[0] if cell.strip()]
+
+            pt = PrettyTable()
+            pt.field_names = headers
+
+            for row in rows[2:]:
+                pt.add_row([cell.strip() for cell in row if cell.strip()])
+
+            return f"\n```\n{pt.get_string()}\n```\n"
+
+        pattern = r"\|.*\|[\n\r]+\|[-:| ]+\|[\n\r]+((?:\|.*\|[\n\r]+)+)"
+        return re.sub(pattern, markdown_to_fixed_width, message)