Reply tracker for Outlook

.cursor/rules/features/reply-tracker.mdc

@@ -22,4 +22,10 @@ When enabling the reply tracker, we create a rule for the user that has the foll
 
 We'll draft a reply for the user automatically if DRAFT_EMAIL is set. The draft will be generated using the email history for this sender as well as the Knowledge base. See `.cursor/rules/features/knowledge.mdc` for more on the Knowledge Base feature.
 
-Enabling `User.outboundReplyTracking` means that when a user sends an email, we'll run an LLM over the email and check if it needs a reply. If it does we'll mark it as Awaiting Reply.
+Typically rules apply to emails as they come in. Rules have no impact afterwards. So to remove a label from a thread later we use the `TRACK_THREAD` action. This action is in charge of removing the Awaiting Reply or To Reply label. Note, adding labels is handled in the usual manner or with `outboundReplyTracking`.
+
+Enabling `EmailAccount.outboundReplyTracking` means that when a user sends an email, we'll run an LLM over the email and check if it needs a reply. If it does we'll mark it as "Awaiting Reply".
+
+Some relevant files:
+- apps/web/utils/reply-tracker/outbound.ts
+- apps/web/utils/reply-tracker/inbound.ts

apps/web/app/api/google/webhook/process-history-item.test.ts

@@ -7,9 +7,8 @@ import { HistoryEventType } from "./types";
 import { ColdEmailSetting } from "@prisma/client";
 import type { gmail_v1 } from "@googleapis/gmail";
 import { isAssistantEmail } from "@/utils/assistant/is-assistant-email";
-import { runColdEmailBlockerWithProvider } from "@/utils/cold-email/is-cold-email";
+import { runColdEmailBlocker } from "@/utils/cold-email/is-cold-email";
 import { blockUnsubscribedEmails } from "@/app/api/google/webhook/block-unsubscribed-emails";
-
 import { markMessageAsProcessing } from "@/utils/redis/message-processing";
 import { GmailLabel } from "@/utils/gmail/label";
 import { categorizeSender } from "@/utils/categorize/senders/categorize";
@@ -53,9 +52,6 @@ vi.mock("@/utils/cold-email/is-cold-email", () => ({
   runColdEmailBlocker: vi
     .fn()
     .mockResolvedValue({ isColdEmail: false, reason: "hasPreviousEmail" }),
-  runColdEmailBlockerWithProvider: vi
-    .fn()
-    .mockResolvedValue({ isColdEmail: false, reason: "hasPreviousEmail" }),
 }));
 vi.mock("@/app/api/google/webhook/block-unsubscribed-emails", () => ({
   blockUnsubscribedEmails: vi.fn().mockResolvedValue(false),
@@ -193,7 +189,7 @@ describe("processHistoryItem", () => {
     await processHistoryItem(createHistoryItem(), options);
 
     expect(blockUnsubscribedEmails).not.toHaveBeenCalled();
-    expect(runColdEmailBlockerWithProvider).not.toHaveBeenCalled();
+    expect(runColdEmailBlocker).not.toHaveBeenCalled();
     expect(processAssistantEmail).toHaveBeenCalledWith({
       message: expect.objectContaining({
         headers: expect.objectContaining({
@@ -235,7 +231,7 @@ describe("processHistoryItem", () => {
     await processHistoryItem(createHistoryItem(), options);
 
     expect(blockUnsubscribedEmails).not.toHaveBeenCalled();
-    expect(runColdEmailBlockerWithProvider).not.toHaveBeenCalled();
+    expect(runColdEmailBlocker).not.toHaveBeenCalled();
   });
 
   it("should skip if email is unsubscribed", async () => {
@@ -247,7 +243,7 @@ describe("processHistoryItem", () => {
     };
     await processHistoryItem(createHistoryItem(), options);
 
-    expect(runColdEmailBlockerWithProvider).not.toHaveBeenCalled();
+    expect(runColdEmailBlocker).not.toHaveBeenCalled();
   });
 
   it("should run cold email blocker when enabled", async () => {
@@ -262,7 +258,7 @@ describe("processHistoryItem", () => {
 
     await processHistoryItem(createHistoryItem(), options);
 
-    expect(runColdEmailBlockerWithProvider).toHaveBeenCalledWith({
+    expect(runColdEmailBlocker).toHaveBeenCalledWith({
       email: expect.objectContaining({
         from: "sender@example.com",
         to: "",
@@ -279,7 +275,7 @@ describe("processHistoryItem", () => {
   });
 
   it("should skip further processing if cold email is detected", async () => {
-    vi.mocked(runColdEmailBlockerWithProvider).mockResolvedValueOnce({
+    vi.mocked(runColdEmailBlocker).mockResolvedValueOnce({
       isColdEmail: true,
       reason: "ai",
       aiReason: "This appears to be a cold email",
@@ -307,7 +303,7 @@ describe("processHistoryItem", () => {
   });
 
   it("should add cold email to digest when coldEmailDigest is true and cold email is detected", async () => {
-    vi.mocked(runColdEmailBlockerWithProvider).mockResolvedValueOnce({
+    vi.mocked(runColdEmailBlocker).mockResolvedValueOnce({
       isColdEmail: true,
       reason: "ai",
       aiReason: "This appears to be a cold email",
@@ -328,7 +324,7 @@ describe("processHistoryItem", () => {
 
     await processHistoryItem(createHistoryItem(), options);
 
-    expect(runColdEmailBlockerWithProvider).toHaveBeenCalledWith({
+    expect(runColdEmailBlocker).toHaveBeenCalledWith({
       email: expect.objectContaining({
         from: "sender@example.com",
         to: "",
@@ -377,13 +373,13 @@ describe("processHistoryItem", () => {
 
     await processHistoryItem(createHistoryItem(), options);
 
-    expect(runColdEmailBlockerWithProvider).not.toHaveBeenCalled();
+    expect(runColdEmailBlocker).not.toHaveBeenCalled();
     expect(categorizeSender).toHaveBeenCalled();
     expect(runRules).toHaveBeenCalled();
   });
 
   it("should process normally when cold email is not detected with coldEmailDigest enabled", async () => {
-    vi.mocked(runColdEmailBlockerWithProvider).mockResolvedValueOnce({
+    vi.mocked(runColdEmailBlocker).mockResolvedValueOnce({
       isColdEmail: false,
       reason: "hasPreviousEmail",
     });
@@ -402,14 +398,14 @@ describe("processHistoryItem", () => {
 
     await processHistoryItem(createHistoryItem(), options);
 
-    expect(runColdEmailBlockerWithProvider).toHaveBeenCalled();
+    expect(runColdEmailBlocker).toHaveBeenCalled();
     expect(categorizeSender).toHaveBeenCalled();
     expect(runRules).toHaveBeenCalled();
   });
 
   it("should add second email from known cold emailer to digest when coldEmailDigest is enabled", async () => {
     // Mock the response for a known cold emailer (already in database)
-    vi.mocked(runColdEmailBlockerWithProvider).mockResolvedValueOnce({
+    vi.mocked(runColdEmailBlocker).mockResolvedValueOnce({
       isColdEmail: true,
       reason: "ai-already-labeled",
       coldEmailId: "existing-cold-email-456", // Existing cold email entry ID
@@ -429,7 +425,7 @@ describe("processHistoryItem", () => {
 
     await processHistoryItem(createHistoryItem("456", "thread-456"), options);
 
-    expect(runColdEmailBlockerWithProvider).toHaveBeenCalledWith({
+    expect(runColdEmailBlocker).toHaveBeenCalledWith({
       email: expect.objectContaining({
         from: "sender@example.com",
         to: "",

apps/web/app/api/google/webhook/process-history-item.ts

@@ -2,7 +2,7 @@ import type { gmail_v1 } from "@googleapis/gmail";
 import prisma from "@/utils/prisma";
 import { emailToContent } from "@/utils/mail";
 import { GmailLabel } from "@/utils/gmail/label";
-import { runColdEmailBlockerWithProvider } from "@/utils/cold-email/is-cold-email";
+import { runColdEmailBlocker } from "@/utils/cold-email/is-cold-email";
 import { runRules } from "@/utils/ai/choose-rule/run-rules";
 import { blockUnsubscribedEmails } from "@/app/api/google/webhook/block-unsubscribed-emails";
 import { categorizeSender } from "@/utils/categorize/senders/categorize";
@@ -24,6 +24,7 @@ import type { ParsedMessage } from "@/utils/types";
 import type { EmailAccountWithAI } from "@/utils/llms/types";
 import { formatError } from "@/utils/error";
 import { createEmailProvider } from "@/utils/email/provider";
+import type { EmailProvider } from "@/utils/email/types";
 import { enqueueDigestItem } from "@/utils/digest/index";
 import { HistoryEventType } from "@/app/api/google/webhook/types";
 import { handleLabelRemovedEvent } from "@/app/api/google/webhook/process-label-removed-event";
@@ -139,7 +140,7 @@ export async function processHistoryItem(
     const isOutbound = parsedMessage.labelIds?.includes(GmailLabel.SENT);
 
     if (isOutbound) {
-      await handleOutbound(emailAccount, parsedMessage, gmail);
+      await handleOutbound(emailAccount, parsedMessage, provider);
       return;
     }
 
@@ -166,7 +167,7 @@ export async function processHistoryItem(
 
       const content = emailToContent(parsedMessage);
 
-      const response = await runColdEmailBlockerWithProvider({
+      const response = await runColdEmailBlocker({
         email: {
           from: parsedMessage.headers.from,
           to: "",
@@ -208,7 +209,7 @@ export async function processHistoryItem(
         select: { category: true },
       });
       if (!existingSender?.category) {
-        await categorizeSender(sender, emailAccount, gmail, accessToken);
+        await categorizeSender(sender, emailAccount, provider);
       }
     }
 
@@ -241,7 +242,7 @@ export async function processHistoryItem(
 async function handleOutbound(
   emailAccount: EmailAccountWithAI,
   message: ParsedMessage,
-  gmail: gmail_v1.Gmail,
+  provider: EmailProvider,
 ) {
   const loggerOptions = {
     email: emailAccount.email,
@@ -257,9 +258,13 @@ async function handleOutbound(
     trackSentDraftStatus({
       emailAccountId: emailAccount.id,
       message,
-      gmail,
+      provider,
+    }),
+    handleOutboundReply({
+      emailAccount,
+      message,
+      provider,
     }),
-    handleOutboundReply({ emailAccount, message, gmail }),
   ]);
 
   if (trackingResult.status === "rejected") {
@@ -282,7 +287,7 @@ async function handleOutbound(
     await cleanupThreadAIDrafts({
       threadId: message.threadId,
       emailAccountId: emailAccount.id,
-      gmail,
+      provider,
     });
   } catch (cleanupError) {
     logger.error("Error during thread draft cleanup", {

apps/web/app/api/outlook/webhook/process-history-item.ts

@@ -1,13 +1,13 @@
 import type { Message } from "@microsoft/microsoft-graph-types";
 import prisma from "@/utils/prisma";
-import { runColdEmailBlockerWithProvider } from "@/utils/cold-email/is-cold-email";
+import { runColdEmailBlocker } from "@/utils/cold-email/is-cold-email";
 import { runRules } from "@/utils/ai/choose-rule/run-rules";
 import { blockUnsubscribedEmails } from "@/app/api/outlook/webhook/block-unsubscribed-emails";
-import { categorizeSenderWithProvider } from "@/utils/categorize/senders/categorize";
+import { categorizeSender } from "@/utils/categorize/senders/categorize";
 import { markMessageAsProcessing } from "@/utils/redis/message-processing";
 import { isAssistantEmail } from "@/utils/assistant/is-assistant-email";
 import { processAssistantEmail } from "@/utils/assistant/process-assistant-email";
-import { handleOutboundReplyWithProvider } from "@/utils/reply-tracker/outbound";
+import { handleOutboundReply } from "@/utils/reply-tracker/outbound";
 import type {
   ProcessHistoryOptions,
   OutlookResourceData,
@@ -16,8 +16,8 @@ import { ColdEmailSetting } from "@prisma/client";
 import { logger } from "@/app/api/outlook/webhook/logger";
 import { extractEmailAddress } from "@/utils/email";
 import {
-  trackSentDraftStatusWithProvider,
-  cleanupThreadAIDraftsWithProvider,
+  trackSentDraftStatus,
+  cleanupThreadAIDrafts,
 } from "@/utils/reply-tracker/draft-tracking";
 import { formatError } from "@/utils/error";
 import { createEmailProvider } from "@/utils/email/provider";
@@ -171,7 +171,7 @@ export async function processHistoryItem(
 
       const content = message.body?.content || "";
 
-      const response = await runColdEmailBlockerWithProvider({
+      const response = await runColdEmailBlocker({
         email: {
           from,
           to: to.join(","),
@@ -205,7 +205,7 @@ export async function processHistoryItem(
         select: { category: true },
       });
       if (!existingSender?.category) {
-        await categorizeSenderWithProvider(sender, emailAccount, emailProvider);
+        await categorizeSender(sender, emailAccount, emailProvider);
       }
     }
 
@@ -297,15 +297,15 @@ async function handleOutbound(
   // Run tracking and outbound reply handling concurrently
   // The individual functions handle their own operational errors.
   const [trackingResult, outboundResult] = await Promise.allSettled([
-    trackSentDraftStatusWithProvider({
+    trackSentDraftStatus({
       emailAccountId: emailAccount.id,
       message: parsedMessage,
-      provider: provider,
+      provider,
     }),
-    handleOutboundReplyWithProvider({
+    handleOutboundReply({
       emailAccount,
       message: parsedMessage,
-      provider: provider,
+      provider,
     }),
   ]);
 
@@ -326,7 +326,7 @@ async function handleOutbound(
   // Run cleanup for any other old/unmodified drafts in the thread
   // Must happen after previous steps
   try {
-    await cleanupThreadAIDraftsWithProvider({
+    await cleanupThreadAIDrafts({
       threadId: conversationId || messageId,
       emailAccountId: emailAccount.id,
       provider: provider,

apps/web/utils/actions/categorize.ts

@@ -2,6 +2,7 @@
 
 import { z } from "zod";
 import { revalidatePath } from "next/cache";
+import { createEmailProvider } from "@/utils/email/provider";
 import {
   type CreateCategoryBody,
   createCategoryBody,
@@ -24,7 +25,6 @@ import { saveCategorizationTotalItems } from "@/utils/redis/categorization-progr
 import { getSenders } from "@/app/api/user/categorize/senders/uncategorized/get-senders";
 import { extractEmailAddress } from "@/utils/email";
 import { actionClient } from "@/utils/actions/safe-action";
-import { getGmailClientForEmail } from "@/utils/account";
 import { prefixPath } from "@/utils/path";
 
 const logger = createScopedLogger("actions/categorize");
@@ -110,21 +110,21 @@ export const categorizeSenderAction = actionClient
   .schema(z.object({ senderAddress: z.string() }))
   .action(
     async ({
-      ctx: { emailAccountId, session },
+      ctx: { emailAccountId, provider },
       parsedInput: { senderAddress },
     }) => {
-      const gmail = await getGmailClientForEmail({ emailAccountId });
-
       const userResult = await validateUserAndAiAccess({ emailAccountId });
       const { emailAccount } = userResult;
 
-      if (!session.session.token) throw new SafeError("No access token");
+      const emailProvider = await createEmailProvider({
+        emailAccountId,
+        provider,
+      });
 
       const result = await categorizeSender(
         senderAddress,
         emailAccount,
-        gmail,
-        session.session.token,
+        emailProvider,
       );
 
       revalidatePath(prefixPath(emailAccountId, "/smart-categories"));