feat: Stip email chain quotes (#454)

control-plane/src/modules/email/index.test.ts

@@ -52,4 +52,53 @@ describe("parseMessage", () => {
     }))).rejects.toThrow("Could not extract clusterId from email address");
   })
 
+  it("should strip trailing email chain quotes", async () => {
+    const result = await parseMessage(buildMessageBody({
+      from: "[email protected]",
+      to: [ `${clusterId}@run.inferable.ai` ],
+      subject: "Subject",
+      body: `Thanks, what is the current time?
+
+> On 2 Jan 2025, at 3:24=E2=80=AFpm, [email protected] wrote:
+
+>=20
+> The following tools are currently available:
+
+
+>=20
+> 1. tavily_search
+>   - Performs web searches using Tavily's AI-powered search API
+>   - Can specify search depth, topic, time range, and result = preferences
+
+>=20
+> 2. myPostgres_getPostgresContext
+>   - Retrieves the database schema information
+>   - No required parameters
+>=20
+
+> 3. myPostgres_executePostgresQuery
+>   - Executes SQL queries
+>   - Requires a valid SQL query string
+>=20
+
+
+> 4. accessKnowledgeArtifacts
+>   - Searches and retrieves knowledge artifacts
+>   - Requires a search query string
+>=20
+
+> 5. currentDateTime
+>   - Gets current date and time in ISO 8601 format and unix timestamp
+>   - No required parameters
+
+
+
+`,
+      messageId: "<[email protected]>"
+    }));
+
+    expect(result).toBeDefined();
+    expect(result.body).toEqual("Thanks, what is the current time?");
+  });
+
 })

control-plane/src/modules/email/index.ts

@@ -7,12 +7,12 @@ import { safeParse } from "../../utilities/safe-parse";
 import { ParsedMail, simpleParser } from "mailparser";
 import { getUserForCluster } from "../clerk";
 import { AuthenticationError, NotFoundError } from "../../utilities/errors";
-import { createRunWithMessage } from "../workflows/workflows";
+import { addMessageAndResume, createRunWithMessage } from "../workflows/workflows";
 import { flagsmith } from "../flagsmith";
 import { InferSelectModel } from "drizzle-orm";
 import { workflowMessages } from "../data";
 import { ses } from "../ses";
-import { getMessageByReference, insertRunMessage, updateMessageReference } from "../workflows/workflow-messages";
+import { getMessageByReference, updateMessageReference } from "../workflows/workflow-messages";
 import { ulid } from "ulid";
 
 const EMAIL_INIT_MESSAGE_ID_META_KEY = "emailInitMessageId";
@@ -172,7 +172,7 @@ export async function parseMessage(message: unknown) {
   }
 
   return {
-    body,
+    body: body ? stripQuoteTail(body) : undefined,
     clusterId,
     ingestionAddresses,
     subject: mail.subject,
@@ -183,6 +183,17 @@ export async function parseMessage(message: unknown) {
   }
 }
 
+// Strip trailing email chain quotes ">"
+export const stripQuoteTail = (message: string) => {
+  const lines = message.split("\n").reverse();
+
+  while (lines[0] && lines[0].startsWith(">") || lines[0].trim() === "") {
+    lines.shift();
+  }
+
+  return lines.reverse().join("\n");
+}
+
 async function handleEmailIngestion(raw: unknown) {
   const message = await parseMessage(raw);
   if (!message.body) {
@@ -323,7 +334,7 @@ const handleExistingChain = async ({
   runId: string;
 }) => {
   logger.info("Continuing existing run from email")
-  await insertRunMessage({
+  await addMessageAndResume({
     id: ulid(),
     clusterId,
     userId,
@@ -332,10 +343,7 @@ const handleExistingChain = async ({
       displayable: {},
       externalReference: messageId,
     },
-    data: {
-      message: body
-    },
+    message: body,
     type: "human",
   })
 }
-