add latest handler updates.

Author: eyw520

servers/fai-lambda/fai-scribe/src/utils/sqs_client.py

@@ -1,5 +1,4 @@
 import json
-import logging
 from typing import Any
 
 import boto3
@@ -73,7 +72,7 @@ def has_interrupt_message(self) -> tuple[bool, str | None]:
         for msg in messages:
             body = msg["body"]
             if body.get("type") == "INTERRUPT":
-                LOGGER.info(f"Found INTERRUPT message in queue")
+                LOGGER.info("Found INTERRUPT message in queue")
                 return True, msg["receipt_handle"]
             else:
                 self._cached_messages.append(msg)
@@ -95,7 +94,10 @@ def get_resume_messages(self) -> list[dict[str, Any]]:
                 resume_messages.append({"body": body, "receipt_handle": msg["receipt_handle"]})
 
         if resume_messages:
-            LOGGER.info(f"Found {len(resume_messages)} RESUME messages ({len([m for m in self._cached_messages if m['body'].get('type') == 'RESUME'])} cached, {len([m for m in messages if m['body'].get('type') == 'RESUME'])} new)")
+            num_resume = len(resume_messages)
+            num_cached = len([m for m in self._cached_messages if m["body"].get("type") == "RESUME"])
+            num_new = len([m for m in messages if m["body"].get("type") == "RESUME"])
+            LOGGER.info(f"Found {num_resume} RESUME messages ({num_cached} cached, {num_new} new)")
 
         self._cached_messages.clear()
 

servers/fai/src/fai/routes/slack.py

@@ -64,7 +64,7 @@
 )
 from fai.utils.slack.response_qa import log_message_for_qa
 
-MESSAGE_CACHE_TTL = 30
+MESSAGE_CACHE_TTL = 600
 
 
 async def cleanup_message_cache() -> None:

servers/fai/src/fai/utils/slack/edit_handler.py

@@ -18,6 +18,10 @@
     VARIABLES,
 )
 from fai.utils.github_utils import get_repo_from_docs_domain
+from fai.utils.sqs_utils import (
+    ResumeMessage,
+    send_message_to_queue,
+)
 
 
 async def get_or_create_editing_session_for_thread(team_id: str, channel_id: str, thread_ts: str) -> str | None:
@@ -112,11 +116,38 @@ async def interrupt_editing_session(editing_id: str) -> bool:
 
 
 async def send_resume_message(editing_id: str, prompts: list[str]) -> bool:
-    # TODO: Implement direct SQS sending from FAI
-    # For now, the Lambda will resume immediately after interruption is detected
-    LOGGER.info(f"Prepared {len(prompts)} prompts for resume (session {editing_id})")
-    LOGGER.debug(f"Prompts: {prompts}")
-    return True
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            response = await client.get(f"{CONFIG.FAI_SERVER_URL}/editing-sessions/{editing_id}")
+
+            if response.status_code != 200:
+                LOGGER.error(f"Failed to fetch session for RESUME message: {response.status_code}")
+                return False
+
+            session_data = response.json()
+            queue_url = session_data["editing_session"].get("queue_url")
+
+            if not queue_url:
+                LOGGER.error(f"No queue_url found for session {editing_id}, cannot send RESUME message")
+                return False
+
+        resume_msg = ResumeMessage(
+            editing_id=editing_id,
+            prompts=prompts,
+            timestamp=datetime.now(UTC).isoformat(),
+        )
+
+        success = await send_message_to_queue(queue_url, resume_msg)
+        if success:
+            LOGGER.info(f"Sent RESUME message with {len(prompts)} prompts to queue for session {editing_id}")
+        else:
+            LOGGER.error(f"Failed to send RESUME message to queue for session {editing_id}")
+
+        return success
+
+    except Exception as e:
+        LOGGER.error(f"Error sending RESUME message for session {editing_id}: {e}", exc_info=True)
+        return False
 
 
 async def create_editing_session(repository: str, base_branch: str = "main") -> str | None:

servers/fai/src/fai/utils/slack/message_handler.py

@@ -384,8 +384,11 @@ async def handle_slack_message(
             LOGGER.info(f"Existing session {existing_editing_id} status: {session_status}")
 
             if session_status and session_status.value == "interrupted":
-                LOGGER.info(f"Session {existing_editing_id} is INTERRUPTED, message already queued")
-                info_text = "⏳ Previous edit is being interrupted, your request will be processed next."
+                LOGGER.info(f"Session {existing_editing_id} is INTERRUPTED, queuing new request via RESUME message")
+
+                await send_resume_message(existing_editing_id, [context.text])
+
+                info_text = "⏳ Previous edit is being interrupted, your request has been queued."
                 return SlackMessageResponse(
                     response_text=info_text,
                     channel=context.channel,
@@ -396,13 +399,11 @@ async def handle_slack_message(
                 )
 
             if session_status and session_status.value in ["startup", "active", "waiting"]:
-                if session_status.value in ["startup", "active"]:
-                    LOGGER.info(
-                        f"Interrupting {session_status.value} session {existing_editing_id} for new edit request"
-                    )
+                if session_status.value == "active":
+                    LOGGER.info(f"Interrupting active session {existing_editing_id} for new edit request")
                     interrupt_success = await interrupt_editing_session(existing_editing_id)
 
-                    if not interrupt_success and session_status.value != "waiting":
+                    if not interrupt_success:
                         LOGGER.error(f"Failed to interrupt session {existing_editing_id}")
                         error_text = "❌ Failed to interrupt the previous edit session. Please try again."
                         return SlackMessageResponse(
@@ -415,24 +416,83 @@ async def handle_slack_message(
                         )
 
                     if integration.slack_bot_token:
-                        if session_status.value == "startup":
-                            interruption_msg = "⚠️ Queued interrupt for starting session - will apply when ready"
-                        else:
-                            interruption_msg = "⚠️ Interrupted previous edit session to process new request"
                         try:
                             await send_slack_message(
                                 channel=context.channel,
-                                text=interruption_msg,
+                                text="⚠️ Interrupted previous edit session to process new request",
                                 bot_token=integration.slack_bot_token,
                                 thread_ts=context.thread_ts,
                             )
                             LOGGER.info(f"Posted interruption message to thread {context.thread_ts}")
                         except Exception as e:
                             LOGGER.error(f"Failed to post interruption message: {e}", exc_info=True)
-                else:
-                    LOGGER.info(f"Session {existing_editing_id} is WAITING and ready to resume")
 
-                await send_resume_message(existing_editing_id, [context.text])
+                    await send_resume_message(existing_editing_id, [context.text])
+
+                elif session_status.value == "startup":
+                    LOGGER.info(f"Interrupting STARTUP session {existing_editing_id} for new edit request")
+                    interrupt_success = await interrupt_editing_session(existing_editing_id)
+
+                    if not interrupt_success:
+                        LOGGER.error(f"Failed to interrupt STARTUP session {existing_editing_id}")
+                        error_text = "❌ Failed to interrupt the previous edit session. Please try again."
+                        return SlackMessageResponse(
+                            response_text=error_text,
+                            channel=context.channel,
+                            thread_ts=context.thread_ts,
+                            bot_token=integration.slack_bot_token,
+                            query_id=None,
+                            user_id=context.user,
+                        )
+
+                    if integration.slack_bot_token:
+                        try:
+                            await send_slack_message(
+                                channel=context.channel,
+                                text="⚠️ Interrupted previous edit session to process new request",
+                                bot_token=integration.slack_bot_token,
+                                thread_ts=context.thread_ts,
+                            )
+                            LOGGER.info(f"Posted interruption message to thread {context.thread_ts}")
+                        except Exception as e:
+                            LOGGER.error(f"Failed to post interruption message: {e}", exc_info=True)
+
+                    await send_resume_message(existing_editing_id, [context.text])
+
+                    return SlackMessageResponse(
+                        response_text="",
+                        channel=context.channel,
+                        thread_ts=context.thread_ts,
+                        bot_token=integration.slack_bot_token,
+                        query_id=None,
+                        user_id=context.user,
+                    )
+
+                elif session_status.value == "waiting":
+                    LOGGER.info(f"Session {existing_editing_id} is WAITING, " f"queuing request without interrupting")
+
+                    await send_resume_message(existing_editing_id, [context.text])
+
+                    if integration.slack_bot_token:
+                        try:
+                            await send_slack_message(
+                                channel=context.channel,
+                                text="⏳ Previous edit is waiting - your request has been queued",
+                                bot_token=integration.slack_bot_token,
+                                thread_ts=context.thread_ts,
+                            )
+                            LOGGER.info(f"Posted queued message to thread {context.thread_ts}")
+                        except Exception as e:
+                            LOGGER.error(f"Failed to post queued message: {e}", exc_info=True)
+
+                    return SlackMessageResponse(
+                        response_text="",
+                        channel=context.channel,
+                        thread_ts=context.thread_ts,
+                        bot_token=integration.slack_bot_token,
+                        query_id=None,
+                        user_id=context.user,
+                    )
 
             else:
                 status_str = session_status.value if session_status else "unknown"