fix(provenance): preserve completions and isolate tool identities (#7630)

Author: icecrasher321

apps/sim/app/api/chat/[identifier]/route.test.ts

@@ -534,6 +534,38 @@ describe('Chat Identifier API Route', () => {
       )
     }, 10000)
 
+    it('projects the internal completion envelope to the public streaming callback', async () => {
+      const response = await POST(createMockNextRequest('POST', { input: 'Hello' }), {
+        params: Promise.resolve({ identifier: 'test-chat' }),
+      })
+      expect(response.status).toBe(200)
+      const onBlockComplete = vi.fn()
+      await vi.mocked(createStreamingResponse).mock.calls[0][0].executeFn({
+        onStream: vi.fn(),
+        onBlockComplete,
+        abortSignal: new AbortController().signal,
+      })
+      await vi.mocked(executeWorkflow).mock.calls[0][4]?.onBlockComplete?.('block-1', {
+        output: { value: 'public output' },
+        outputBlockId: 'child:block-1',
+        resolvedSecretTraceProvenance: {
+          version: 1,
+          complete: true,
+          entries: [{ encryptedValue: 'private-ciphertext' }],
+          scope: { userId: 'user-1', workspaceId: 'workspace-1' },
+        },
+        executionTime: 1,
+        executionOrder: 0,
+        startedAt: '2026-01-01T00:00:00Z',
+        endedAt: '2026-01-01T00:00:01Z',
+      })
+      expect(onBlockComplete).toHaveBeenCalledExactlyOnceWith(
+        'block-1',
+        { value: 'public output' },
+        'child:block-1'
+      )
+    })
+
     it('executes with the email proven by the chat authentication gate', async () => {
       mockValidateChatAuth.mockResolvedValueOnce({
         authorized: true,

apps/sim/app/api/chat/[identifier]/route.ts

@@ -414,7 +414,8 @@ export const POST = withRouteHandler(
                 isSecureMode: true,
                 workflowTriggerType: 'chat',
                 onStream,
-                onBlockComplete,
+                onBlockComplete: (blockId, data) =>
+                  onBlockComplete(blockId, data.output, data.outputBlockId),
                 skipLoggingComplete: true,
                 abortSignal,
                 executionMode: 'stream',

apps/sim/app/api/workflows/[id]/execute/route.ts

@@ -1718,7 +1718,8 @@ async function handleExecutePost(
               isSecureMode: false,
               workflowTriggerType: triggerType === 'chat' ? 'chat' : 'api',
               onStream,
-              onBlockComplete,
+              onBlockComplete: (blockId, data) =>
+                onBlockComplete(blockId, data.output, data.outputBlockId),
               skipLoggingComplete: true,
               includeFileBase64,
               base64MaxBytes,

apps/sim/background/resume-execution.ts

@@ -14,6 +14,7 @@ import {
   getTimeoutErrorMessage,
 } from '@/lib/core/execution-limits'
 import { withCascadeLock } from '@/lib/table/cascade-lock'
+import type { WorkflowCellProgressWriter } from '@/lib/table/cell-write'
 import { isExecCancelled } from '@/lib/table/deps'
 import type { RowExecutionMetadata } from '@/lib/table/types'
 import { classifyWorkflowCellTerminalResult } from '@/lib/table/workflow-cell-result'
@@ -247,7 +248,7 @@ function throwIfResumeAttemptTimedOut(
 }
 
 type CellWriters = {
-  cellOnBlockComplete: (blockId: string, output: unknown) => Promise<void>
+  cellOnBlockComplete: WorkflowCellProgressWriter['onBlockComplete']
   writeCellTerminal: (
     status: 'completed' | 'error' | 'cancelled' | 'paused',
     error: string | null

apps/sim/lib/copilot/async-runs/errors.ts

@@ -0,0 +1,7 @@
+/** A durable tool identity must never be reused by a different run. */
+export class AsyncToolCallOwnershipError extends Error {
+  constructor() {
+    super('Async tool call belongs to another run')
+    this.name = 'AsyncToolCallOwnershipError'
+  }
+}

apps/sim/lib/copilot/async-runs/repository.test.ts

@@ -406,4 +406,28 @@ describe('async tool repository single-row semantics', () => {
       expect(dbChainMockFns.values).not.toHaveBeenCalled()
     }
   )
+
+  it('refuses a provider-ID collision with an existing row in another run', async () => {
+    dbChainMockFns.limit.mockResolvedValueOnce([
+      { runId: 'old-run', toolCallId: 'provider-shared-call', toolName: 'browser_close_tab' },
+    ])
+    await expect(
+      upsertAsyncToolCall({
+        runId: 'current-run',
+        toolCallId: 'provider-shared-call',
+        toolName: 'glob',
+      })
+    ).rejects.toThrow('Async tool call belongs to another run')
+    expect(dbChainMockFns.values).not.toHaveBeenCalled()
+  })
+
+  it('refuses a foreign row that wins the insert race', async () => {
+    dbChainMockFns.limit
+      .mockResolvedValueOnce([])
+      .mockResolvedValueOnce([{ runId: 'other-run', toolCallId: 'call-race' }])
+    dbChainMockFns.returning.mockResolvedValueOnce([])
+    await expect(
+      upsertAsyncToolCall({ runId: 'current-run', toolCallId: 'call-race', toolName: 'glob' })
+    ).rejects.toThrow('Async tool call belongs to another run')
+  })
 })

apps/sim/lib/copilot/async-runs/repository.ts

@@ -11,6 +11,7 @@ import { createLogger } from '@sim/logger'
 import { filterUndefined } from '@sim/utils/object'
 import { sanitizeValueForJsonb } from '@sim/utils/string'
 import { and, desc, eq, inArray, isNull, or, sql } from 'drizzle-orm'
+import { AsyncToolCallOwnershipError } from '@/lib/copilot/async-runs/errors'
 import { TraceAttr } from '@/lib/copilot/generated/trace-attributes-v1'
 import { TraceSpan } from '@/lib/copilot/generated/trace-spans-v1'
 import { markSpanForError } from '@/lib/copilot/request/otel'
@@ -219,7 +220,12 @@ export async function upsertAsyncToolCall(input: {
     },
     async () => {
       const existing = await getAsyncToolCall(input.toolCallId)
-      if (existing) return existing
+      if (existing) {
+        if (input.runId && existing.runId !== input.runId) {
+          throw new AsyncToolCallOwnershipError()
+        }
+        return existing
+      }
 
       const incomingStatus = input.status ?? 'pending'
       const effectiveRunId = input.runId ?? null
@@ -250,7 +256,11 @@ export async function upsertAsyncToolCall(input: {
         .onConflictDoNothing()
         .returning()
 
-      return row ?? getAsyncToolCall(input.toolCallId)
+      const persisted = row ?? (await getAsyncToolCall(input.toolCallId))
+      if (persisted && persisted.runId !== effectiveRunId) {
+        throw new AsyncToolCallOwnershipError()
+      }
+      return persisted
     }
   )
 }