Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions apps/sim/app/api/chat/[identifier]/route.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
3 changes: 2 additions & 1 deletion apps/sim/app/api/chat/[identifier]/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand Down
3 changes: 2 additions & 1 deletion apps/sim/app/api/workflows/[id]/execute/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
3 changes: 2 additions & 1 deletion apps/sim/background/resume-execution.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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
Expand Down
7 changes: 7 additions & 0 deletions apps/sim/lib/copilot/async-runs/errors.ts
Original file line number Diff line number Diff line change
@@ -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'
}
}
24 changes: 24 additions & 0 deletions apps/sim/lib/copilot/async-runs/repository.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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')
})
})
14 changes: 12 additions & 2 deletions apps/sim/lib/copilot/async-runs/repository.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
}
)
}
Expand Down
Loading
Loading