Add [THREAD_REPLY] support: DraftResponseAgent results posted as thread replies to customer inquiry messages

Author: songmeo

src/bro/agents/email_agent.py

@@ -386,6 +386,7 @@ def _handle_customer_inquiry(self, state: EmailAgentState) -> EmailAgentState:
             "customer": customer_display,
             "question": extracted["question"],
             "order_details": order_details,
+            "message_id": email.get("message_ids", [""])[0] if email.get("message_ids") else "",
         }
 
         state["customer_inquiries"].append(inquiry_data)
@@ -432,6 +433,7 @@ def _generate_report(self, state: EmailAgentState) -> EmailAgentState:
                     report.append(f"Customer: {inquiry['customer']}")
                     report.append(f"Question: {inquiry['question']}")
                     report.append(f"Order Details: {inquiry['order_details']}")
+                    report.append(f"Message ID: {inquiry['message_id']}")
 
         final_report = "\n".join(report)
         state["messages"].append(AIMessage(content=final_report))

src/bro/conversation.py

@@ -380,7 +380,13 @@ def _on_task_completed_cb(self, message: str, scheduled: bool = False) -> None:
                 _logger.info("Scheduled task completion: posting to main channel without threading")
 
         try:
+            last_message_ts = None  # Track the last posted message for thread replies
             for msg in messages:
+                # Check if this is a thread reply
+                is_thread_reply = msg.startswith("[THREAD_REPLY]")
+                if is_thread_reply:
+                    msg = msg.replace("[THREAD_REPLY]\n", "").strip()
+
                 input_data = textwrap.dedent(
                     f"""\
                 via:  
@@ -398,12 +404,33 @@ def _on_task_completed_cb(self, message: str, scheduled: bool = False) -> None:
                     }
                 ]
 
-                _logger.info(f"Requesting conversation response...")
-                conversation_response = self._request_inference(self._context)
-                output = conversation_response["output"]
-                if not output:
-                    _logger.warning("No output from conversation model; response: %s", conversation_response)
-                self._process_response_output(output)
+                # If this is a thread reply, temporarily set thread_ts to last message
+                if is_thread_reply and last_message_ts:
+                    with self._thread_ts_lock:
+                        saved_for_reply = self._current_thread_ts
+                        self._current_thread_ts = last_message_ts
+
+                    _logger.info(f"Posting as thread reply to {last_message_ts}")
+                    conversation_response = self._request_inference(self._context)
+                    output = conversation_response["output"]
+                    if not output:
+                        _logger.warning("No output from conversation model; response: %s", conversation_response)
+                    self._process_response_output(output)
+
+                    # Restore thread_ts
+                    with self._thread_ts_lock:
+                        self._current_thread_ts = saved_for_reply
+                else:
+                    _logger.info(f"Requesting conversation response...")
+                    conversation_response = self._request_inference(self._context)
+                    output = conversation_response["output"]
+                    if not output:
+                        _logger.warning("No output from conversation model; response: %s", conversation_response)
+
+                    # Get the message ts from the response to use for next thread reply
+                    # This is a simplification - ideally we'd extract it from the actual Slack post
+                    self._process_response_output(output)
+                    last_message_ts = self._current_thread_ts  # Store for potential thread reply
         finally:
             # Restore the original thread_ts
             with self._thread_ts_lock:

src/bro/reasoner/openai_generic.py

@@ -1013,34 +1013,53 @@ def _process(self, item: dict[str, Any]) -> tuple[
                             _logger.info("Running EmailAgent...")
                             result = self._email_agent.run()
 
-                            # Auto-trigger DraftResponseAgent for customer inquiries
+                            # Process each inquiry with DraftResponseAgent and append as thread replies
                             if self._draft_response_agent and result:
                                 # Extract full inquiry context from the result
                                 import re
 
                                 # Split by ===SPLIT_MESSAGE=== to get individual inquiries
                                 inquiries = result.split("===SPLIT_MESSAGE===")
+                                summary = inquiries[0] if inquiries else ""
+                                processed_parts = [summary]
+
                                 for inquiry in inquiries[1:]:  # Skip the summary part
                                     inquiry = inquiry.strip()
                                     if inquiry.startswith("Customer:"):
                                         try:
-                                            # Extract customer, question, and order details
+                                            # Extract customer, question, order details, and message_id
                                             customer_match = re.search(r"^Customer:\s*(.+)$", inquiry, re.MULTILINE)
                                             question_match = re.search(r"^Question:\s*(.+)$", inquiry, re.MULTILINE)
                                             order_match = re.search(r"^Order Details:\s*(.+)$", inquiry, re.MULTILINE)
+                                            message_id_match = re.search(r"^Message ID:\s*(.+)$", inquiry, re.MULTILINE)
 
                                             if customer_match:
                                                 customer = customer_match.group(1)
                                                 question = question_match.group(1) if question_match else ""
                                                 order_details = order_match.group(1) if order_match else ""
+                                                message_id = message_id_match.group(1) if message_id_match else ""
 
-                                                # Pass full context to DraftResponseAgent
-                                                inquiry_context = f"Customer: {customer}\nQuestion: {question}\nOrder: {order_details}"
+                                                # Run DraftResponseAgent
+                                                inquiry_context = f"Customer: {customer}\nQuestion: {question}\nOrder: {order_details}\nMessage ID: {message_id}"
                                                 _logger.info(f"Running DraftResponseAgent for: {customer}")
                                                 draft_result = self._draft_response_agent.run(inquiry_context)
                                                 _logger.info(f"DraftResponseAgent result: {draft_result}")
+
+                                                # Add inquiry message
+                                                processed_parts.append(f"===SPLIT_MESSAGE===\n{inquiry}")
+                                                # Add draft as a thread reply (marked with special delimiter)
+                                                processed_parts.append(
+                                                    f"===SPLIT_MESSAGE===\n[THREAD_REPLY]\n{draft_result}"
+                                                )
                                         except Exception as e:
                                             _logger.error(f"DraftResponseAgent failed: {e}")
+                                            # Keep original inquiry even if draft fails
+                                            processed_parts.append(f"===SPLIT_MESSAGE===\n{inquiry}")
+                                    else:
+                                        processed_parts.append(f"===SPLIT_MESSAGE===\n{inquiry}")
+
+                                # Rebuild result with drafts as thread replies
+                                result = "\n".join(processed_parts)
                         else:
                             result = "EmailAgent not available. Google Workspace client is required."