diff --git a/blueprints/conversation/__init__.py b/blueprints/conversation/__init__.py index 13309c8..0eb932a 100644 --- a/blueprints/conversation/__init__.py +++ b/blueprints/conversation/__init__.py @@ -15,6 +15,7 @@ from quart_jwt_extended import ( import blueprints.users.models from metrics import count_result, observe_response, time_response from utils.image_process import image_content_block +from utils.strip_markdown import extract_agent_text from utils.image_upload import ImageValidationError, process_image from utils.s3_client import generate_presigned_url as s3_presigned_url from utils.s3_client import get_image as s3_get_image @@ -171,7 +172,7 @@ async def query(): } response = await main_agent.ainvoke(payload, config=agent_config) - message = response.get("messages", [])[-1].content + message = extract_agent_text(response.get("messages", [])[-1].content) await persist_agent_tool_messages( conversation, user, response.get("messages", []) ) diff --git a/blueprints/conversation/logic.py b/blueprints/conversation/logic.py index 91f2e31..f17c0e9 100644 --- a/blueprints/conversation/logic.py +++ b/blueprints/conversation/logic.py @@ -54,6 +54,7 @@ async def add_message_to_conversation( speaker: str, user: blueprints.users.models.User, image_key: str | None = None, + message_handle: str | None = None, ) -> ConversationMessage: # Bump updated_at on every message so conversations sort by last activity. # `update_fields` bypasses the auto_now recompute, so set it explicitly. @@ -72,6 +73,7 @@ async def add_message_to_conversation( speaker=speaker, conversation=conversation, image_key=image_key, + message_handle=message_handle, ) return message @@ -105,7 +107,9 @@ async def get_conversation_for_user(user: blueprints.users.models.User) -> Conve async def get_conversation_for_channel( user: blueprints.users.models.User, channel: str ) -> Conversation: - conversation = await Conversation.filter(user=user, channel=channel).first() + conversation = await ( + Conversation.filter(user=user, channel=channel).order_by("-updated_at").first() + ) if conversation is None: conversation = await Conversation.create( name=f"{user.username}'s {channel} chat", user=user, channel=channel diff --git a/blueprints/conversation/models.py b/blueprints/conversation/models.py index 0b1fd15..309bc84 100644 --- a/blueprints/conversation/models.py +++ b/blueprints/conversation/models.py @@ -37,6 +37,7 @@ class ConversationMessage(Model): created_at = fields.DatetimeField(auto_now_add=True) speaker = fields.CharEnumField(enum_type=Speaker, max_length=10) image_key = fields.CharField(max_length=512, null=True, default=None) + message_handle = fields.CharField(max_length=255, null=True, default=None) class Meta: table = "conversation_messages" diff --git a/blueprints/imessage/__init__.py b/blueprints/imessage/__init__.py index ba7cf4e..0852927 100644 --- a/blueprints/imessage/__init__.py +++ b/blueprints/imessage/__init__.py @@ -14,9 +14,10 @@ from blueprints.conversation.logic import ( add_message_to_conversation, persist_agent_tool_messages, ) -from blueprints.conversation.models import Speaker +from blueprints.conversation.models import Conversation, ConversationMessage, Speaker from blueprints.conversation.agents import main_agent from blueprints.conversation.memory import build_system_prompt_with_memories +from utils.strip_markdown import extract_agent_text, strip_markdown imessage_blueprint = Blueprint("imessage_api", __name__, url_prefix="/api/imessage") @@ -59,25 +60,60 @@ def _check_rate_limit(phone_number: str) -> bool: return True -async def send_imessage(to_number: str, content: str) -> dict: +async def send_imessage( + to_number: str, + content: str, + reply_to_handle: str | None = None, +) -> dict: """Send an iMessage via SendBlue API.""" from_number = os.getenv("SENDBLUE_FROM_NUMBER", "") + payload: dict = { + "number": to_number, + "from_number": from_number, + "content": content, + } + if reply_to_handle: + payload["reply_to"] = {"message_handle": reply_to_handle} + async with httpx.AsyncClient() as client: resp = await client.post( f"{SENDBLUE_API_BASE}/api/send-message", headers=_get_sendblue_headers(), - json={ - "number": to_number, - "from_number": from_number, - "content": content, - }, + json=payload, timeout=30, ) resp.raise_for_status() return resp.json() +async def send_typing_indicator( + to_number: str, + state: str = "start", + max_duration_ms: int | None = None, +) -> None: + """Send or stop a typing indicator. Errors are logged and swallowed.""" + from_number = os.getenv("SENDBLUE_FROM_NUMBER", "") + payload: dict = { + "number": to_number, + "from_number": from_number, + "state": state, + } + if max_duration_ms is not None: + payload["max_duration_ms"] = max_duration_ms + + try: + async with httpx.AsyncClient() as client: + await client.post( + f"{SENDBLUE_API_BASE}/api/send-typing-indicator", + headers=_get_sendblue_headers(), + json=payload, + timeout=10, + ) + except Exception as e: + logger.warning(f"Failed to send typing indicator ({state}): {e}") + + def validate_sendblue_signature(f): """Decorator to validate the SendBlue webhook signing secret.""" @@ -114,6 +150,11 @@ async def webhook(): media_url = data.get("media_url") is_outbound = data.get("is_outbound", False) + # Threading fields from SendBlue + inbound_handle = data.get("message_handle") + reply_to_handle = (data.get("reply_to") or {}).get("message_handle") + originator_handle = (data.get("thread_originator") or {}).get("message_handle") + # Ignore outbound messages (our own replies echoed back) if is_outbound: return jsonify({"status": "ignored"}), 200 @@ -183,23 +224,73 @@ async def webhook(): await send_imessage(from_number, "Sorry, this feature is restricted to admins.") return jsonify({"status": "forbidden"}), 200 - # Get or create conversation + # /threads command — list recent iMessage conversations + if content.strip().lower() == "/threads": + threads = ( + await Conversation.filter(user=user, channel="imessage") + .order_by("-updated_at") + .limit(5) + ) + if not threads: + await send_imessage(from_number, "No threads found.") + else: + lines = ["Your recent threads (most recent first):"] + for i, t in enumerate(threads, 1): + updated = t.updated_at.strftime("%b %-d") + lines.append(f"{i}. {t.name} ({updated})") + lines.append("\nReply to any message to continue its thread.") + await send_imessage(from_number, "\n".join(lines)) + return jsonify({"status": "ok"}), 200 + + # Determine conversation based on iMessage reply threading: + # - Reply to an existing message → continue that conversation + # - New standalone message → start a fresh conversation try: - conversation = await get_conversation_for_channel(user=user, channel="imessage") + if reply_to_handle: + lookup_handle = originator_handle or reply_to_handle + parent_msg = await ConversationMessage.get_or_none( + message_handle=lookup_handle, + conversation__user=user, + conversation__channel="imessage", + ) + if parent_msg: + conversation = await Conversation.get(id=parent_msg.conversation_id) + logger.info( + f"Continuing conversation {conversation.id} via reply to {lookup_handle!r}" + ) + else: + # Handle not found (pre-migration messages) — fall back to most recent + logger.warning( + f"Reply handle {lookup_handle!r} not in DB; falling back to most recent conversation" + ) + conversation = await get_conversation_for_channel( + user=user, channel="imessage" + ) + else: + conversation = await Conversation.create( + name="New iMessage", + user=user, + channel="imessage", + ) + logger.info( + f"New iMessage conversation {conversation.id} for {user.username}" + ) + await conversation.fetch_related("messages") except Exception as e: - logger.error(f"Failed to get conversation for user {user.username}: {e}") + logger.error(f"Failed to get/create conversation for {user.username}: {e}") await send_imessage( from_number, "Sorry, something went wrong. Please try again later." ) return jsonify({"status": "error"}), 200 - # Add user message to conversation + # Add user message (store the inbound handle for future reply lookups) await add_message_to_conversation( conversation=conversation, message=content, speaker="user", user=user, + message_handle=inbound_handle, ) # Attach image directly to the agent request if present @@ -216,7 +307,11 @@ async def webhook(): logger.error(f"Failed to load iMessage image: {e}") content = f"{content or ''}\n\n[An image was sent but could not be loaded.]".strip() + # Show typing indicator for up to 2 minutes while the agent processes + await send_typing_indicator(from_number, state="start", max_duration_ms=120_000) + # Build messages payload for LangChain agent + response_text = "Sorry, I'm having trouble thinking right now." try: messages = await conversation.messages.all() recent_messages = list(messages)[-20:] @@ -244,25 +339,30 @@ async def webhook(): response = await main_agent.ainvoke( {"messages": messages_payload}, config=agent_config ) - response_text = response.get("messages", [])[-1].content + response_text = extract_agent_text(response.get("messages", [])[-1].content) await persist_agent_tool_messages( conversation, user, response.get("messages", []) ) except Exception as e: logger.error(f"Error invoking agent: {e}") - response_text = "Sorry, I'm having trouble thinking right now." + finally: + await send_typing_indicator(from_number, state="stop") + + # Send reply threaded under the user's inbound message + send_resp = await send_imessage( + from_number, + strip_markdown(response_text), + reply_to_handle=inbound_handle, + ) + outbound_handle = (send_resp or {}).get("message_handle") - # Save and send response await add_message_to_conversation( conversation=conversation, message=response_text, speaker="simba", user=user, + message_handle=outbound_handle, ) - from utils.strip_markdown import strip_markdown - - await send_imessage(from_number, strip_markdown(response_text)) - return jsonify({"status": "ok"}), 200 diff --git a/blueprints/whatsapp/__init__.py b/blueprints/whatsapp/__init__.py index 30faeea..e438eb4 100644 --- a/blueprints/whatsapp/__init__.py +++ b/blueprints/whatsapp/__init__.py @@ -16,6 +16,7 @@ from blueprints.conversation.logic import ( from blueprints.conversation.models import Speaker from blueprints.conversation.agents import main_agent from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT +from utils.strip_markdown import extract_agent_text whatsapp_blueprint = Blueprint("whatsapp_api", __name__, url_prefix="/api/whatsapp") @@ -198,7 +199,7 @@ async def webhook(): # Invoke LangChain agent logger.info(f"Invoking LangChain agent with {len(messages_payload)} messages") response = await main_agent.ainvoke({"messages": messages_payload}) - response_text = response.get("messages", [])[-1].content + response_text = extract_agent_text(response.get("messages", [])[-1].content) await persist_agent_tool_messages( conversation, user, response.get("messages", []) ) diff --git a/utils/strip_markdown.py b/utils/strip_markdown.py index 8094d5a..2b73045 100644 --- a/utils/strip_markdown.py +++ b/utils/strip_markdown.py @@ -1,5 +1,26 @@ import re +_THINK_RE = re.compile(r".*?", re.DOTALL) + + +def extract_agent_text(content) -> str: + """Extract plain text from a LangChain AIMessage content. + + Handles both str and list-of-blocks content (the latter occurs with some + tool-using or vision models), and strips ... tokens emitted + by reasoning models like Qwen3. + """ + if isinstance(content, list): + text = "".join( + block.get("text", "") + if isinstance(block, dict) and block.get("type") != "thinking" + else (str(block) if not isinstance(block, dict) else "") + for block in content + ) + else: + text = str(content) if content is not None else "" + return _THINK_RE.sub("", text).strip() + def strip_markdown(text: str) -> str: """Strip markdown formatting from text for plain-text channels like iMessage."""