Fix user insertion conflicts and add message parts tracking

- Add onConflictDoNothing to user creation to handle race conditions
- Fetch conflicted users that were prevented from insertion
- Create part records for agent task messages to properly track content

Author: caelinsutch

apps/interaction-worker/src/services/loop/store-loop-messages.ts

@@ -96,11 +96,30 @@ export const storeLoopMessages = async (
       const newUsers = await db
         .insert(users)
         .values(missingUsers.map((phoneNumber) => ({ phoneNumber })))
+        .onConflictDoNothing({ target: users.phoneNumber })
         .returning();
 
       newUsers.forEach((user) => {
         userMap.set(user.phoneNumber, user.id);
       });
+
+      // If onConflictDoNothing prevented some inserts, fetch those users
+      if (newUsers.length < missingUsers.length) {
+        const stillMissingPhones = missingUsers.filter(
+          (phone) => !userMap.has(phone),
+        );
+        if (stillMissingPhones.length > 0) {
+          const conflictedUsers = await db
+            .select()
+            .from(users)
+            .where(inArray(users.phoneNumber, stillMissingPhones));
+
+          conflictedUsers.forEach((user) => {
+            userMap.set(user.phoneNumber, user.id);
+          });
+        }
+      }
+
       logger.info("Created new users", {
         count: newUsers.length,
         phoneNumbers: missingUsers,

apps/interaction-worker/src/tools/send-message-to-agent.ts

@@ -1,5 +1,5 @@
 import type { Agent, getDb } from "@poppy/db";
-import { messages as messagesTable } from "@poppy/db";
+import { messages as messagesTable, parts } from "@poppy/db";
 import { logger } from "@poppy/hono-helpers";
 import { generateId, tool } from "ai";
 import { z } from "zod";
@@ -101,8 +101,9 @@ The agent has tools for a wide variety of tasks. Use this tool often.
         });
 
       // Record the message in the messages table
+      const taskMessageId = generateId();
       await db.insert(messagesTable).values({
-        id: generateId(),
+        id: taskMessageId,
         conversationId,
         fromAgentId: interactionAgentId,
         toAgentId: executionAgent.id,
@@ -114,6 +115,18 @@ The agent has tools for a wide variety of tasks. Use this tool often.
         },
       });
 
+      // Create part record for the message content
+      await db.insert(parts).values({
+        id: generateId(),
+        messageId: taskMessageId,
+        type: "text",
+        content: {
+          type: "text",
+          text: message,
+        },
+        order: 0,
+      });
+
       // Call execution-worker via RPC to execute the task
       logger
         .withTags({