fix(conversations): Keep the root actor when later turns record execution

packages/junior/src/chat/conversations/sql/store.ts

@@ -510,6 +510,11 @@ export class SqlStore implements ConversationStore {
       const execution = incomingIsFresh
         ? args.execution
         : (existing?.execution ?? args.execution);
+      // Root Actor and origin Source come from the saved Conversation, as in
+      // recordActivity. A later Turn, such as an event on a web Conversation,
+      // must not derive a system Actor from its own Source.
+      const actor = mergeActor(existing?.actor, args.actor);
+      const source = existing?.source ?? args.source;
       await this.upsertConversation({
         conversation: {
           schemaVersion: 1,
@@ -523,8 +528,8 @@ export class SqlStore implements ConversationStore {
           ...(existing?.location ? { location: existing.location } : undefined),
           ...(args.channelName ? { channelName: args.channelName } : undefined),
           ...(args.destination ? { destination: args.destination } : undefined),
-          ...(args.actor ? { actor: args.actor } : undefined),
-          ...(args.source ? { source: args.source } : undefined),
+          ...(actor ? { actor } : undefined),
+          ...(source ? { source } : undefined),
           ...(args.title ? { title: args.title } : undefined),
           ...(args.visibility ? { visibility: args.visibility } : undefined),
           execution,

packages/junior/tests/component/conversation-sql-store.test.ts

@@ -646,7 +646,7 @@ describe("conversation SQL store", () => {
     }
   }, 15_000);
 
-  it("fills missing actor identity from later trusted profile observations", async () => {
+  it("fills missing actor identity from later observations and keeps it across event writes", async () => {
     const fixture = await createJuniorSqlFixture();
 
     try {
@@ -677,6 +677,17 @@ describe("conversation SQL store", () => {
         source: "slack",
         nowMs: 2_000,
       });
+      // An event write without an Actor, such as a mailbox mirror after the
+      // Redis work state expired, must not replace the Slack root Actor.
+      await store.recordExecution({
+        conversationId: CONVERSATION_ID,
+        createdAtMs: 1_000,
+        execution: { status: "idle", updatedAtMs: 3_000 },
+        lastActivityAtMs: 3_000,
+        metrics: null,
+        source: "event",
+        updatedAtMs: 3_000,
+      });
 
       await expect(store.listByActivity({ limit: 5 })).resolves.toMatchObject([
         {
@@ -686,6 +697,7 @@ describe("conversation SQL store", () => {
             fullName: "Casey Example",
             slackUserId: "U123",
           },
+          source: "slack",
         },
       ]);
     } finally {

packages/junior/tests/integration/conversation-turn-work.test.ts

@@ -638,6 +638,9 @@ describe("Conversation mailbox Turn work", () => {
       source,
       state: "completed",
     });
+    const root = await conversationStore.get({ conversationId });
+    // The event Turn keeps the web root Actor.
+    expect(root?.actor).toMatchObject({ email: actor.email });
     const transcript = (
       await getConversationEventStore().loadHistory(conversationId)
     ).flatMap((event) =>