fix (tasks): triggered tasks composio setup

Author: Kabeer2004

src/server/main/integrations/routes.py

@@ -420,27 +420,33 @@ async def composio_webhook(request: Request):
     """
     try:
         payload = await request.json()
-        user_id = payload.get("userId")
-        trigger_slug = payload.get("triggerSlug")
-        event_data = payload.get("payload")
+        # Correctly parse the payload based on the provided sample structure
+        event_data = payload.get("data")
+        if not event_data:
+            raise HTTPException(status_code=400, detail="Webhook payload missing 'data' object.")
 
-        if not all([user_id, trigger_slug, event_data]):
-            raise HTTPException(status_code=400, detail="Missing required fields in webhook payload.")
+        user_id = event_data.get("user_id")
+        trigger_type = payload.get("type")
 
+        if not all([user_id, trigger_type, event_data]):
+            raise HTTPException(status_code=400, detail="Missing required fields (user_id, type) in webhook payload.")
+
+        # Map from Composio's trigger type to our internal service_name and event_type
+        # Using lowercase to match the sample payload
         service_name_map = {
-            "GOOGLECALENDAR_GOOGLE_CALENDAR_EVENT_SYNC_TRIGGER": "gcalendar",
-            "GMAIL_NEW_GMAIL_MESSAGE": "gmail"
+            "googlecalendar_google_calendar_event_sync_trigger": "gcalendar",
+            "gmail_new_gmail_message": "gmail"
         }
         event_type_map = {
-            "GOOGLECALENDAR_GOOGLE_CALENDAR_EVENT_SYNC_TRIGGER": "new_event",
-            "GMAIL_NEW_GMAIL_MESSAGE": "new_email"
+            "googlecalendar_google_calendar_event_sync_trigger": "new_event",
+            "gmail_new_gmail_message": "new_email"
         }
 
-        service_name = service_name_map.get(trigger_slug)
-        event_type = event_type_map.get(trigger_slug)
+        service_name = service_name_map.get(trigger_type)
+        event_type = event_type_map.get(trigger_type)
 
         if not service_name:
-            logger.warning(f"Received webhook for unhandled trigger slug: {trigger_slug}")
+            logger.warning(f"Received webhook for unhandled trigger type: {trigger_type}")
             return JSONResponse(content={"status": "ignored", "reason": "unhandled trigger"})
 
         logger.info(f"Received Composio trigger for user '{user_id}' - Service: '{service_name}', Event: '{event_type}'")
@@ -461,7 +467,7 @@ async def composio_webhook(request: Request):
             logger.info(f"Event for user '{user_id}' was discarded by the pre-filter.")
             return JSONResponse(content={"status": "ignored", "reason": "pre-filter discard"})
 
-        # 3. Dispatch to Celery worker
+        # 3. Dispatch to a Celery task that handles finding and running all matching triggered workflows.
         execute_triggered_task.delay(
             user_id=user_id,
             source=service_name,

src/server/main/tasks/prompts.py

@@ -27,6 +27,7 @@
         - `source`: The service that triggers the workflow (e.g., "gmail", "gcalendar").
         - `event`: The specific event (e.g., "new_email", "new_event").
         - `filter`: A dictionary of conditions to match (e.g., `{{"from": "boss@example.com"}}`).
+      The task will execute *after* the trigger occurs, using the event data (like the email content) as context. 
     - CRUCIAL DISTINCTION: Differentiate between the *task's execution time* (`run_at`) and the *event's time* mentioned in the prompt. A task to arrange a future event (e.g., 'book a flight for next month', 'schedule a meeting for Friday') should be executed *now* to make the arrangement. Therefore, its `run_at` should be null, since setting run_at to null makes the task run immediately. The future date belongs in the task `description`.
     - Ambiguity: Phrases like "weekly hourly" are ambiguous. Interpret "weekly" as the frequency and ignore "hourly".
     - Use the current time and user's timezone to resolve relative dates like "tomorrow", "next Friday at 2pm", etc. correctly.

src/server/workers/planner/prompts.py

@@ -15,6 +15,7 @@
 6.  Anticipate Information Gaps: If crucial information is still missing after checking context, the first step should be to use a tool to find it (e.g., `internet_search` for public information, `gpeople` for contacts, `memory` for personal information, `gcalendar` for upcoming events and so on).
 7.  Output a Clear Plan: Your final output must be a single, valid JSON object containing a concise description of the overall goal and a list of specific, actionable steps for the executor.
 8. If the task is scheduled or recurring, only plan for an indivisual occurrence, not the entire series. The executor will handle scheduling. For example, if the user asks you to "Send a news summary every day at 8 AM", your plan should only include the steps for the indivisual run such as "Search for the news", "Summarize the news", "Send the news on WhatsApp". The executor will then handle the scheduling for future occurrences. Do NOT include any steps using the `gcalendar` tool to create a recurring or scheduled events.
+9. If the task is triggered by an event (like a new email or calendar event), your plan should focus on the immediate actions to take after that event has occured, such as "Search for the email", "Extract relevant information", "Send a summary to the user". The executor will handle the triggering mechanism. The executor is receiving the incoming email or calendar event as input, along with your plan so write each plan step as if the event has already occured and you are now processing it. For example, if the triggered task is to send a summary for each new email, your plan should be to directly read the email and summarize it, you do not need to search for the email as it is part of the input trigger.
 
 Here is the complete list of services (tools) available to the executor agent, that you can use in your plan:
 {{