fix(chat): log only turns this call finalized, include recovered turns, and resolve the email at send time

Author: waleedlatif1

apps/sim/lib/mothership/chat/chat-log.ts

@@ -1,19 +1,21 @@
+import { db } from '@sim/db'
+import { user } from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { getErrorMessage } from '@sim/utils/errors'
+import { eq } from 'drizzle-orm'
 import { env } from '@/lib/core/config/env'
 import type { OrchestratorResult } from '@/lib/mothership/request/types'
 
 const logger = createLogger('ChatLog')
 
 const CHAT_LOG_TIMEOUT_MS = 5_000
 
-/** The finished turn's identity, captured when the user's message is admitted. */
+/** The turn a Chat send admitted — rebuilt identically when a relay pod recovers it. */
 export interface ChatTurnLogContext {
   chatId: string
   messageId: string
   requestId: string
   userId: string
-  userEmail?: string
   userMessage: string
   mode: 'assistant' | 'agent' | 'plan'
   startedAt: number
@@ -34,6 +36,28 @@ export function logChatTurn(
 ): void {
   const url = env.SIM_LOGGING_WORKFLOW_URL
   if (!url) return
+  const durationMs = Date.now() - context.startedAt
+  void sendChatTurn(url, context, result, status, durationMs).catch((error) => {
+    logger.warn('Chat log workflow request failed', {
+      chatId: context.chatId,
+      requestId: context.requestId,
+      error: getErrorMessage(error),
+    })
+  })
+}
+
+async function sendChatTurn(
+  url: string,
+  context: ChatTurnLogContext,
+  result: OrchestratorResult,
+  status: ChatTurnStatus,
+  durationMs: number
+): Promise<void> {
+  const [row] = await db
+    .select({ email: user.email })
+    .from(user)
+    .where(eq(user.id, context.userId))
+    .limit(1)
 
   const headers: Record<string, string> = { 'Content-Type': 'application/json' }
   if (env.SIM_LOGGING_WORKFLOW_API_KEY) headers['X-API-Key'] = env.SIM_LOGGING_WORKFLOW_API_KEY
@@ -45,7 +69,7 @@ export function logChatTurn(
     chatId: context.chatId,
     messageId: context.messageId,
     userId: context.userId,
-    userEmail: context.userEmail,
+    userEmail: row?.email,
     userMessage: context.userMessage,
     assistantResponse: result.content,
     status,
@@ -55,34 +79,25 @@ export function logChatTurn(
     mode: context.mode,
     source: 'workspace-chat',
     startedAt: new Date(context.startedAt).toISOString(),
-    durationMs: Date.now() - context.startedAt,
+    durationMs,
     usage: {
       inputTokens: result.usage?.prompt ?? 0,
       outputTokens: result.usage?.completion ?? 0,
     },
   }
 
-  void fetch(url, {
+  const response = await fetch(url, {
     method: 'POST',
     headers,
     body: JSON.stringify({ input }),
     signal: AbortSignal.timeout(CHAT_LOG_TIMEOUT_MS),
   })
-    .then(async (response) => {
-      await response.body?.cancel()
-      if (!response.ok) {
-        logger.warn('Chat log workflow returned a non-2xx status', {
-          status: response.status,
-          chatId: context.chatId,
-          requestId: context.requestId,
-        })
-      }
-    })
-    .catch((error) => {
-      logger.warn('Chat log workflow request failed', {
-        chatId: context.chatId,
-        requestId: context.requestId,
-        error: getErrorMessage(error),
-      })
+  await response.body?.cancel()
+  if (!response.ok) {
+    logger.warn('Chat log workflow returned a non-2xx status', {
+      status: response.status,
+      chatId: context.chatId,
+      requestId: context.requestId,
     })
+  }
 }

apps/sim/lib/mothership/chat/completion.ts

@@ -24,7 +24,7 @@ export function buildOnComplete(params: {
   organizationId?: string
   userId?: string
   requestMode?: 'assistant' | 'agent' | 'plan'
-  /** Present for Chat turns that feed the operator chat log. */
+  /** Present for workspace and organization Chat turns, which feed the operator chat log. */
   chatLog?: ChatTurnLogContext
   /**
    * Root agent span for this request. When present, the final
@@ -102,7 +102,7 @@ export function buildOnComplete(params: {
       const assistantMessage = buildPersistedAssistantMessage(result, requestId, params.requestMode)
       const hasPartial =
         !!assistantMessage.content?.trim() || (assistantMessage.contentBlocks?.length ?? 0) > 0
-      await finalizeAssistantTurn({
+      const finalization = await finalizeAssistantTurn({
         runController,
         chatId,
         userMessageId,
@@ -112,7 +112,9 @@ export function buildOnComplete(params: {
         ...(result.success ? {} : { streamMarkerPolicy: 'active-or-cleared' as const }),
       })
 
-      if (chatLog) logChatTurn(chatLog, result, result.success ? 'success' : 'error')
+      if (chatLog && finalization.updated) {
+        logChatTurn(chatLog, result, result.success ? 'success' : 'error')
+      }
       if (notifyChatStatus) {
         publishChatStatusChanged(
           { workspaceId, organizationId, userId },
@@ -144,7 +146,7 @@ export function buildOnError(params: {
   organizationId?: string
   userId?: string
   requestMode?: 'assistant' | 'agent' | 'plan'
-  /** Present for Chat turns that feed the operator chat log. */
+  /** Present for workspace and organization Chat turns, which feed the operator chat log. */
   chatLog?: ChatTurnLogContext
 }) {
   const {
@@ -176,15 +178,15 @@ export function buildOnError(params: {
         error: result?.error || getErrorMessage(error),
       }
       const assistantMessage = buildPersistedAssistantMessage(failed, requestId, params.requestMode)
-      await finalizeAssistantTurn({
+      const finalization = await finalizeAssistantTurn({
         runController,
         chatId,
         userMessageId,
         assistantMessage,
         streamMarkerPolicy: 'active-or-cleared',
       })
 
-      if (chatLog) logChatTurn(chatLog, failed, 'error')
+      if (chatLog && finalization.updated) logChatTurn(chatLog, failed, 'error')
       if (notifyChatStatus) {
         publishChatStatusChanged(
           { workspaceId, organizationId, userId },

apps/sim/lib/mothership/chat/post.ts

@@ -986,7 +986,6 @@ export async function handleUnifiedChatPost(req: NextRequest) {
   let requestId = ''
   const executionId = generateId()
   const runId = generateId()
-  const startedAt = Date.now()
 
   try {
     const session = await getSession()
@@ -1489,18 +1488,16 @@ export async function handleUnifiedChatPost(req: NextRequest) {
       }
       const requestMode =
         body.mode === 'plan' ? 'plan' : body.mode === 'assistant' ? 'assistant' : 'agent'
-      /** Workspace and organization Chat feed the operator chat log; the workflow panel does not. */
       const chatLog: ChatTurnLogContext | undefined =
         branch.kind !== 'workflow' && actualChatId
           ? {
               chatId: actualChatId,
               messageId: userMessageId,
               requestId,
               userId: authenticatedUserId,
-              ...(authenticatedUserEmail ? { userEmail: authenticatedUserEmail } : {}),
               userMessage: body.message,
               mode: requestMode,
-              startedAt,
+              startedAt: Date.now(),
             }
           : undefined
       const stream = createSSEStream({

apps/sim/lib/mothership/request/application/recover-stream.test.ts

@@ -73,6 +73,7 @@ const run = {
   workspaceId: '33333333-3333-4333-8333-333333333333',
   status: 'active',
   workflowId: null,
+  startedAt: new Date('2026-01-01T00:00:00Z'),
   requestContext: {
     requestId: 'request',
     controllerToken: 'old-controller',

apps/sim/lib/mothership/request/application/recover-stream.ts

@@ -13,6 +13,7 @@ import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { getLatestRunForStream } from '@/lib/mothership/async-runs/repository'
 import { defineAuthorizedChatUseCase } from '@/lib/mothership/chat/application/authorized-chat-use-case'
 import { resolveOwnedChatContext } from '@/lib/mothership/chat/application/context'
+import type { ChatTurnLogContext } from '@/lib/mothership/chat/chat-log'
 import { buildOnComplete, buildOnError } from '@/lib/mothership/chat/completion'
 import { restoreBillingAdmission } from '@/lib/mothership/request/lifecycle/admission'
 import { claimRunController } from '@/lib/mothership/request/lifecycle/controller-ownership'
@@ -121,21 +122,31 @@ export const readChatStream = defineAuthorizedChatUseCase({
       if (workspaceId && !userPermission)
         throw new OrchestrationError('forbidden', 'Workspace access revoked')
       const requestId = typeof saved?.requestId === 'string' ? saved.requestId : generateId()
+      const requestMode: ChatTurnLogContext['mode'] =
+        intent.mode === 'plan' ? 'plan' : intent.mode === 'assistant' ? 'assistant' : 'agent'
       const completion = {
         chatId,
         userMessageId: run.streamId,
         requestId,
         workspaceId,
         organizationId,
         userId,
-        requestMode:
-          intent.mode === 'plan'
-            ? ('plan' as const)
-            : intent.mode === 'assistant'
-              ? ('assistant' as const)
-              : ('agent' as const),
+        requestMode,
         notifyWorkspaceStatus: true,
         runController: { id: run.id, token: lease.value },
+        ...(config.data.goRoute === '/api/mothership'
+          ? {
+              chatLog: {
+                chatId,
+                messageId: run.streamId,
+                requestId,
+                userId,
+                userMessage: intent.message,
+                mode: requestMode,
+                startedAt: run.startedAt.getTime(),
+              },
+            }
+          : {}),
       }
       const stream = createSSEStream({
         userId,