Fix merge conflicts

Author: Sg312

apps/sim/lib/copilot/orchestrator/index.ts

@@ -1,12 +1,17 @@
 import { createLogger } from '@sim/logger'
 import { updateRunStatus } from '@/lib/copilot/async-runs/repository'
 import { SIM_AGENT_API_URL, SIM_AGENT_VERSION } from '@/lib/copilot/constants'
-import { prepareExecutionContext } from '@/lib/copilot/orchestrator/tool-executor'
-import type {
-  ExecutionContext,
-  OrchestratorOptions,
-  OrchestratorResult,
-  SSEEvent,
+import {
+  isToolAvailableOnSimSide,
+  prepareExecutionContext,
+} from '@/lib/copilot/orchestrator/tool-executor'
+import {
+  type ExecutionContext,
+  isTerminalToolCallStatus,
+  type OrchestratorOptions,
+  type OrchestratorResult,
+  type SSEEvent,
+  type ToolCallState,
 } from '@/lib/copilot/orchestrator/types'
 import { env } from '@/lib/core/config/env'
 import { getEffectiveDecryptedEnv } from '@/lib/environment/utils'
@@ -19,6 +24,7 @@ const RESUME_UPSTREAM_MAX_ATTEMPTS = 3
 const RESUME_UPSTREAM_RETRY_MS = 500
 const ASYNC_RESUME_DIAG_TAG = '[ASYNC_RESUME_DIAG]'
 
+<<<<<<< HEAD
 interface CheckpointReadyResponse {
   success?: boolean
   checkpointId?: string
@@ -31,6 +37,15 @@ interface CheckpointReadyResponse {
   code?: string
   retryable?: boolean
 }
+=======
+function didAsyncToolSucceed(input: {
+  durableStatus?: string | null
+  durableResult?: Record<string, unknown>
+  durableError?: string | null
+  toolStateStatus?: string | undefined
+}) {
+  const { durableStatus, durableResult, durableError, toolStateStatus } = input
+>>>>>>> 0c80438ed (fix(mothership): async resume and tool result ordering (#3735))
 
 function sleep(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms))
@@ -122,12 +137,27 @@ async function waitForCheckpointReady(
     }
   }
 
+<<<<<<< HEAD
   return {
     checkpointId,
     ready: false,
     error: 'Checkpoint did not become ready in time',
     retryable: true,
   }
+=======
+  if (toolStateStatus === 'success') return true
+  if (toolStateStatus === 'error' || toolStateStatus === 'cancelled') return false
+
+  return false
+}
+
+interface ReadyContinuationTool {
+  toolCallId: string
+  toolState?: ToolCallState
+  durableRow?: Awaited<ReturnType<typeof getAsyncToolCall>>
+  needsDurableClaim: boolean
+  alreadyClaimedByWorker: boolean
+>>>>>>> 0c80438ed (fix(mothership): async resume and tool result ordering (#3735))
 }
 
 export interface OrchestrateStreamOptions extends OrchestratorOptions {
@@ -277,6 +307,7 @@ export async function orchestrateCopilotStream(
       if (!continuation) break
 
       let resumeReady = false
+<<<<<<< HEAD
       const localPendingPromises = continuation.pendingToolCallIds
         .map((toolCallId) => context.pendingToolPromises.get(toolCallId))
         .filter(
@@ -298,6 +329,227 @@ export async function orchestrateCopilotStream(
         await Promise.allSettled(localPendingPromises)
         logger.warn(ASYNC_RESUME_DIAG_TAG, {
           phase: 'local_async_tools_settled',
+=======
+      let resumeRetries = 0
+      for (;;) {
+        claimedToolCallIds = []
+        claimedByWorkerId = null
+        const resumeWorkerId = continuation.runId || context.runId || context.messageId
+        const readyTools: ReadyContinuationTool[] = []
+        const localPendingPromises: Promise<unknown>[] = []
+        const missingToolCallIds: string[] = []
+
+        for (const toolCallId of continuation.pendingToolCallIds) {
+          const durableRow = await getAsyncToolCall(toolCallId).catch(() => null)
+          const localPendingPromise = context.pendingToolPromises.get(toolCallId)
+          const toolState = context.toolCalls.get(toolCallId)
+
+          if (localPendingPromise) {
+            localPendingPromises.push(localPendingPromise)
+            logger.info('Waiting for local async tool completion before retrying resume claim', {
+              toolCallId,
+              runId: continuation.runId,
+            })
+            continue
+          }
+
+          if (durableRow && isTerminalAsyncStatus(durableRow.status)) {
+            if (durableRow.claimedBy && durableRow.claimedBy !== resumeWorkerId) {
+              missingToolCallIds.push(toolCallId)
+              logger.warn('Async tool continuation is waiting on a claim held by another worker', {
+                toolCallId,
+                runId: continuation.runId,
+                claimedBy: durableRow.claimedBy,
+              })
+              continue
+            }
+            readyTools.push({
+              toolCallId,
+              toolState,
+              durableRow,
+              needsDurableClaim: durableRow.claimedBy !== resumeWorkerId,
+              alreadyClaimedByWorker: durableRow.claimedBy === resumeWorkerId,
+            })
+            continue
+          }
+
+          if (
+            !durableRow &&
+            toolState &&
+            isTerminalToolCallStatus(toolState.status) &&
+            !isToolAvailableOnSimSide(toolState.name)
+          ) {
+            logger.info('Including Go-handled tool in resume payload (no Sim-side row)', {
+              toolCallId,
+              toolName: toolState.name,
+              status: toolState.status,
+              runId: continuation.runId,
+            })
+            readyTools.push({
+              toolCallId,
+              toolState,
+              needsDurableClaim: false,
+              alreadyClaimedByWorker: false,
+            })
+            continue
+          }
+
+          logger.warn('Skipping already-claimed or missing async tool resume', {
+            toolCallId,
+            runId: continuation.runId,
+            durableStatus: durableRow?.status,
+            toolStateStatus: toolState?.status,
+          })
+          missingToolCallIds.push(toolCallId)
+        }
+
+        if (localPendingPromises.length > 0) {
+          await Promise.allSettled(localPendingPromises)
+          continue
+        }
+
+        if (missingToolCallIds.length > 0) {
+          if (resumeRetries < 3) {
+            resumeRetries++
+            logger.info('Retrying async resume after some tool calls were not yet ready', {
+              checkpointId: continuation.checkpointId,
+              runId: continuation.runId,
+              retry: resumeRetries,
+              missingToolCallIds,
+            })
+            await new Promise((resolve) => setTimeout(resolve, 250 * resumeRetries))
+            continue
+          }
+          throw new Error(
+            `Failed to resume async tool continuation: pending tool calls were not ready (${missingToolCallIds.join(', ')})`
+          )
+        }
+
+        if (readyTools.length === 0) {
+          if (resumeRetries < 3 && continuation.pendingToolCallIds.length > 0) {
+            resumeRetries++
+            logger.info('Retrying async resume because no tool calls were ready yet', {
+              checkpointId: continuation.checkpointId,
+              runId: continuation.runId,
+              retry: resumeRetries,
+            })
+            await new Promise((resolve) => setTimeout(resolve, 250 * resumeRetries))
+            continue
+          }
+          throw new Error('Failed to resume async tool continuation: no tool calls were ready')
+        }
+
+        const claimCandidates = readyTools.filter((tool) => tool.needsDurableClaim)
+        const newlyClaimedToolCallIds: string[] = []
+        const claimFailures: string[] = []
+
+        for (const tool of claimCandidates) {
+          const claimed = await claimCompletedAsyncToolCall(tool.toolCallId, resumeWorkerId).catch(
+            () => null
+          )
+          if (!claimed) {
+            claimFailures.push(tool.toolCallId)
+            continue
+          }
+          newlyClaimedToolCallIds.push(tool.toolCallId)
+        }
+
+        if (claimFailures.length > 0) {
+          if (newlyClaimedToolCallIds.length > 0) {
+            logger.info('Releasing async tool claims after claim contention during resume', {
+              checkpointId: continuation.checkpointId,
+              runId: continuation.runId,
+              newlyClaimedToolCallIds,
+              claimFailures,
+            })
+            await Promise.all(
+              newlyClaimedToolCallIds.map((toolCallId) =>
+                releaseCompletedAsyncToolClaim(toolCallId, resumeWorkerId).catch(() => null)
+              )
+            )
+          }
+          if (resumeRetries < 3) {
+            resumeRetries++
+            logger.info('Retrying async resume after claim contention', {
+              checkpointId: continuation.checkpointId,
+              runId: continuation.runId,
+              retry: resumeRetries,
+              claimFailures,
+            })
+            await new Promise((resolve) => setTimeout(resolve, 250 * resumeRetries))
+            continue
+          }
+          throw new Error(
+            `Failed to resume async tool continuation: unable to claim tool calls (${claimFailures.join(', ')})`
+          )
+        }
+
+        claimedToolCallIds = [
+          ...readyTools
+            .filter((tool) => tool.alreadyClaimedByWorker)
+            .map((tool) => tool.toolCallId),
+          ...newlyClaimedToolCallIds,
+        ]
+        claimedByWorkerId = claimedToolCallIds.length > 0 ? resumeWorkerId : null
+
+        logger.info('Resuming async tool continuation', {
+          checkpointId: continuation.checkpointId,
+          runId: continuation.runId,
+          toolCallIds: readyTools.map((tool) => tool.toolCallId),
+        })
+
+        const durableRows = await getAsyncToolCalls(
+          readyTools.map((tool) => tool.toolCallId)
+        ).catch(() => [])
+        const durableByToolCallId = new Map(durableRows.map((row) => [row.toolCallId, row]))
+
+        const results = await Promise.all(
+          readyTools.map(async (tool) => {
+            const durable = durableByToolCallId.get(tool.toolCallId) || tool.durableRow
+            const durableStatus = durable?.status
+            const durableResult =
+              durable?.result && typeof durable.result === 'object'
+                ? (durable.result as Record<string, unknown>)
+                : undefined
+            const success = didAsyncToolSucceed({
+              durableStatus,
+              durableResult,
+              durableError: durable?.error,
+              toolStateStatus: tool.toolState?.status,
+            })
+            const data =
+              durableResult ||
+              (tool.toolState?.result?.output as Record<string, unknown> | undefined) ||
+              (success
+                ? { message: 'Tool completed' }
+                : {
+                    error: durable?.error || tool.toolState?.error || 'Tool failed',
+                  })
+
+            if (
+              durableStatus &&
+              !isTerminalAsyncStatus(durableStatus) &&
+              !isDeliveredAsyncStatus(durableStatus)
+            ) {
+              logger.warn('Async tool row was claimed for resume without terminal durable state', {
+                toolCallId: tool.toolCallId,
+                status: durableStatus,
+              })
+            }
+
+            return {
+              callId: tool.toolCallId,
+              name: durable?.toolName || tool.toolState?.name || '',
+              data,
+              success,
+            }
+          })
+        )
+
+        context.awaitingAsyncContinuation = undefined
+        route = '/api/tools/resume'
+        payload = {
+>>>>>>> 0c80438ed (fix(mothership): async resume and tool result ordering (#3735))
           checkpointId: continuation.checkpointId,
           runId: continuation.runId,
           pendingToolCallIds: continuation.pendingToolCallIds,

apps/sim/lib/copilot/orchestrator/sse/handlers/handlers.test.ts

@@ -242,4 +242,76 @@ describe('sse-handlers tool lifecycle', () => {
     expect(markToolComplete).toHaveBeenCalledTimes(1)
     expect(context.toolCalls.get('tool-upsert-fail')?.status).toBe('success')
   })
+
+  it('does not execute a tool if a terminal tool_result arrives before local execution starts', async () => {
+    let resolveUpsert: ((value: null) => void) | undefined
+    upsertAsyncToolCall.mockImplementationOnce(
+      () =>
+        new Promise((resolve) => {
+          resolveUpsert = resolve
+        })
+    )
+    const onEvent = vi.fn()
+
+    await sseHandlers.tool_call(
+      {
+        type: 'tool_call',
+        data: { id: 'tool-race', name: 'read', arguments: { workflowId: 'workflow-1' } },
+      } as any,
+      context,
+      execContext,
+      { onEvent, interactive: false, timeout: 1000 }
+    )
+
+    await sseHandlers.tool_result(
+      {
+        type: 'tool_result',
+        toolCallId: 'tool-race',
+        data: { id: 'tool-race', success: true, result: { ok: true } },
+      } as any,
+      context,
+      execContext,
+      { onEvent, interactive: false, timeout: 1000 }
+    )
+
+    resolveUpsert?.(null)
+    await new Promise((resolve) => setTimeout(resolve, 0))
+
+    expect(executeToolServerSide).not.toHaveBeenCalled()
+    expect(markToolComplete).not.toHaveBeenCalled()
+    expect(context.toolCalls.get('tool-race')?.status).toBe('success')
+    expect(context.toolCalls.get('tool-race')?.result?.output).toEqual({ ok: true })
+  })
+
+  it('does not execute a tool if a tool_result arrives before the tool_call event', async () => {
+    const onEvent = vi.fn()
+
+    await sseHandlers.tool_result(
+      {
+        type: 'tool_result',
+        toolCallId: 'tool-early-result',
+        toolName: 'read',
+        data: { id: 'tool-early-result', name: 'read', success: true, result: { ok: true } },
+      } as any,
+      context,
+      execContext,
+      { onEvent, interactive: false, timeout: 1000 }
+    )
+
+    await sseHandlers.tool_call(
+      {
+        type: 'tool_call',
+        data: { id: 'tool-early-result', name: 'read', arguments: { workflowId: 'workflow-1' } },
+      } as any,
+      context,
+      execContext,
+      { onEvent, interactive: false, timeout: 1000 }
+    )
+
+    await new Promise((resolve) => setTimeout(resolve, 0))
+
+    expect(executeToolServerSide).not.toHaveBeenCalled()
+    expect(markToolComplete).not.toHaveBeenCalled()
+    expect(context.toolCalls.get('tool-early-result')?.status).toBe('success')
+  })
 })