From b5bae34edf2b5e4272c44e1fa761c8e78019ddda Mon Sep 17 00:00:00 2001 From: Siddharth Ganesan Date: Wed, 12 Aug 2026 15:45:09 -0700 Subject: [PATCH] Make async tool resume delivery recoverable --- .../[id]/execute/route.async.test.ts | 38 +++- .../app/api/workflows/[id]/execute/route.ts | 60 ++--- .../utils/workflow-execution-utils.ts | 5 +- apps/sim/lib/copilot/generated/metrics-v1.ts | 2 + .../lib/copilot/generated/tool-catalog-v1.ts | 15 +- .../lib/copilot/generated/tool-schemas-v1.ts | 15 +- .../generated/trace-attribute-values-v1.ts | 10 + .../copilot/generated/trace-attributes-v1.ts | 2 + .../copilot/request/handlers/handlers.test.ts | 132 +++++++++++ apps/sim/lib/copilot/request/handlers/tool.ts | 55 +++++ .../sim/lib/copilot/request/handlers/types.ts | 32 ++- .../lifecycle/resume-leg-context.test.ts | 35 +++ .../lib/copilot/request/lifecycle/run.test.ts | 154 +++++++++++++ apps/sim/lib/copilot/request/lifecycle/run.ts | 215 +++++++++++++----- apps/sim/lib/copilot/request/metrics.ts | 15 ++ .../tools/workflow-client-fallback.test.ts | 16 +- .../request/tools/workflow-client-fallback.ts | 3 + .../tools/client/run-tool-execution.test.ts | 32 +++ .../tools/client/run-tool-execution.ts | 13 +- .../lib/copilot/tools/workflow-tools.test.ts | 86 +++++++ apps/sim/lib/copilot/tools/workflow-tools.ts | 101 ++++++++ 21 files changed, 910 insertions(+), 126 deletions(-) create mode 100644 apps/sim/lib/copilot/tools/workflow-tools.test.ts diff --git a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts b/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts index 4adac12ff2..20bf4ff7e5 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts +++ b/apps/sim/app/api/workflows/[id]/execute/route.async.test.ts @@ -852,8 +852,13 @@ describe('workflow execute async route', () => { status: 'pending', }, { id: 'copilot-run-1', userId: 'session-user-1', workflowId: 'workflow-1' }, + 403, + 'COPILOT_WORKFLOW_TOOL_BINDING_AWAITING_APPROVAL', ], [ + // A finished call is a benign duplicate, not a defect: some other runner + // already owns this tool call, so it reports the same conflict the + // execution claim does and the client stays silent. 'terminal tool row', { toolCallId: 'copilot-tool-1', @@ -863,6 +868,8 @@ describe('workflow execute async route', () => { status: 'completed', }, { id: 'copilot-run-1', userId: 'session-user-1', workflowId: 'workflow-1' }, + 409, + 'COPILOT_WORKFLOW_EXECUTION_CONFLICT', ], [ 'different workflow target', @@ -874,6 +881,8 @@ describe('workflow execute async route', () => { status: 'running', }, { id: 'copilot-run-1', userId: 'session-user-1', workflowId: 'workflow-1' }, + 403, + 'COPILOT_WORKFLOW_TOOL_BINDING_WORKFLOW_MISMATCH', ], [ 'different execution actor', @@ -885,19 +894,28 @@ describe('workflow execute async route', () => { status: 'running', }, { id: 'copilot-run-1', userId: 'other-user', workflowId: 'workflow-1' }, + 403, + 'COPILOT_WORKFLOW_TOOL_BINDING_FOREIGN_OWNER', ], - ])('rejects a Copilot binding owned by a %s', async (_caseName, toolCall, run) => { - mockGetAsyncToolCall.mockResolvedValueOnce(toolCall) - mockGetRunSegment.mockResolvedValueOnce(run) + ['missing tool row', null, null, 404, 'COPILOT_WORKFLOW_TOOL_BINDING_UNKNOWN'], + ])( + 'rejects a Copilot binding owned by a %s', + async (_caseName, toolCall, run, expectedStatus, expectedCode) => { + mockGetAsyncToolCall.mockResolvedValueOnce(toolCall) + mockGetRunSegment.mockResolvedValueOnce(run) - const response = await POST(createBoundCopilotExecutionRequest(), { - params: Promise.resolve({ id: 'workflow-1' }), - }) + const response = await POST(createBoundCopilotExecutionRequest(), { + params: Promise.resolve({ id: 'workflow-1' }), + }) - expect(response.status).toBe(403) - expect(mockExecuteWorkflowCore).not.toHaveBeenCalled() - expect(loggingSessionMockFns.mockSetTrustedExecutionCorrelation).not.toHaveBeenCalled() - }) + expect(response.status).toBe(expectedStatus) + // The reason must be machine-readable — an opaque 403 is what stopped the + // client telling a benign duplicate from a real failure. + await expect(response.json()).resolves.toMatchObject({ code: expectedCode }) + expect(mockExecuteWorkflowCore).not.toHaveBeenCalled() + expect(loggingSessionMockFns.mockSetTrustedExecutionCorrelation).not.toHaveBeenCalled() + } + ) it('rejects Copilot workflow bindings outside the interactive SSE surface', async () => { const response = await POST(createBoundCopilotExecutionRequest({ stream: false }), { diff --git a/apps/sim/app/api/workflows/[id]/execute/route.ts b/apps/sim/app/api/workflows/[id]/execute/route.ts index 25704a1030..ba44c86f0c 100644 --- a/apps/sim/app/api/workflows/[id]/execute/route.ts +++ b/apps/sim/app/api/workflows/[id]/execute/route.ts @@ -21,7 +21,6 @@ import { type BillingAttributionSnapshot, requireBillingAttributionHeader, } from '@/lib/billing/core/billing-attribution' -import { isWorkflowToolExecutionClaimable } from '@/lib/copilot/async-runs/lifecycle' import { claimWorkflowToolExecution, getAsyncToolCall, @@ -29,10 +28,12 @@ import { releaseWorkflowToolExecutionClaim, } from '@/lib/copilot/async-runs/repository' import { COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE } from '@/lib/copilot/constants' +import { CopilotDegradedReason } from '@/lib/copilot/generated/trace-attribute-values-v1' +import { recordDegraded } from '@/lib/copilot/request/metrics' import { ASYNC_WORKFLOW_DEPLOYMENT_ERRORS, - isWorkflowToolName, - resolveWorkflowToolTargetId, + type CopilotWorkflowToolBindingResult, + classifyWorkflowToolBinding, } from '@/lib/copilot/tools/workflow-tools' import { admissionRejectedResponse, tryAdmit } from '@/lib/core/admission/gate' import { @@ -168,25 +169,19 @@ const SERVER_EXECUTION_ID_CLAIM_ATTEMPTS = 3 export const runtime = 'nodejs' export const dynamic = 'force-dynamic' -async function isValidCopilotWorkflowToolBinding(params: { +async function resolveCopilotWorkflowToolBinding(params: { toolCallId: string userId: string workflowId: string -}): Promise { +}): Promise { const toolCall = await getAsyncToolCall(params.toolCallId) - if ( - !toolCall || - !isWorkflowToolName(toolCall.toolName) || - !isWorkflowToolExecutionClaimable(toolCall.status, toolCall.permissionDecision) - ) { - return false - } - - const run = await getRunSegment(toolCall.runId) - return ( - run?.userId === params.userId && - resolveWorkflowToolTargetId(toolCall.args, run.workflowId) === params.workflowId - ) + const run = toolCall ? await getRunSegment(toolCall.runId) : null + return classifyWorkflowToolBinding({ + toolCall, + run, + userId: params.userId, + workflowId: params.workflowId, + }) } function createExecutionJsonResponse( @@ -1019,18 +1014,29 @@ async function handleExecutePost( ) } - if ( - copilotToolCallId && - !(await isValidCopilotWorkflowToolBinding({ + if (copilotToolCallId) { + const binding = await resolveCopilotWorkflowToolBinding({ toolCallId: copilotToolCallId, userId, workflowId, - })) - ) { - return NextResponse.json( - { error: 'Copilot workflow tool binding was not found' }, - { status: 403 } - ) + }) + if (!binding.ok) { + // This rejection happens before any LoggingSession exists, so it leaves + // no execution log and no workflow span — log the reason or it is + // invisible everywhere except the browser console. + // This rejection happens before a LoggingSession or any workflow span + // exists, so the counter is the only place it becomes visible. + recordDegraded(CopilotDegradedReason.BindingRejected) + reqLogger.warn('Rejected Copilot workflow tool execution', { + copilotToolCallId, + workflowId, + reason: binding.rejection.code, + }) + return NextResponse.json( + { error: binding.rejection.message, code: binding.rejection.code }, + { status: binding.rejection.statusCode } + ) + } } if (inputFromExecutionId) { diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts index 3f888b2e19..4dc59961db 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/utils/workflow-execution-utils.ts @@ -1070,7 +1070,10 @@ export async function executeWorkflowWithFullLogging( error: errorMessage, httpStatus: response.status, }) - throw new Error(errorMessage) + // Keep the status and code on the thrown error. Downgrading to a bare Error + // discarded both, so callers could not tell a Copilot binding rejection from + // any other 4xx — and the reason never reached the agent that could fix it. + throw new ExecutionStreamHttpError(errorMessage, response.status, errorCode) } if (!response.body) { diff --git a/apps/sim/lib/copilot/generated/metrics-v1.ts b/apps/sim/lib/copilot/generated/metrics-v1.ts index 06c74839c4..b379803f21 100644 --- a/apps/sim/lib/copilot/generated/metrics-v1.ts +++ b/apps/sim/lib/copilot/generated/metrics-v1.ts @@ -19,6 +19,7 @@ export const Metric = { CopilotCacheWrite: 'copilot.cache.write', CopilotChatBlobBytes: 'copilot.chat.blob.bytes', CopilotChatBlobCount: 'copilot.chat.blob.count', + CopilotDegradedCount: 'copilot.degraded.count', CopilotFileReadDuration: 'copilot.file.read.duration', CopilotFileReadSize: 'copilot.file.read.size', CopilotMessagesSerializeDuration: 'copilot.messages.serialize.duration', @@ -48,6 +49,7 @@ export const MetricValues: readonly MetricValue[] = [ 'copilot.cache.write', 'copilot.chat.blob.bytes', 'copilot.chat.blob.count', + 'copilot.degraded.count', 'copilot.file.read.duration', 'copilot.file.read.size', 'copilot.messages.serialize.duration', diff --git a/apps/sim/lib/copilot/generated/tool-catalog-v1.ts b/apps/sim/lib/copilot/generated/tool-catalog-v1.ts index 805bdf29f8..91fe4c4c2e 100644 --- a/apps/sim/lib/copilot/generated/tool-catalog-v1.ts +++ b/apps/sim/lib/copilot/generated/tool-catalog-v1.ts @@ -4268,14 +4268,14 @@ export const RunBlock: ToolCatalogEntry = { workflowId: { type: 'string', description: - 'Optional workflow ID to run. If not provided, uses the current workflow in context.', + 'ID of the workflow to run. Always pass this explicitly — outside a workflow-scoped chat there is no current workflow to fall back to, and the run is rejected without it.', }, workflow_input: { type: 'object', description: 'JSON object with key-value mappings where each key is an input field name', }, }, - required: ['blockId'], + required: ['workflowId', 'blockId'], }, clientExecutable: true, } @@ -4396,14 +4396,14 @@ export const RunFromBlock: ToolCatalogEntry = { workflowId: { type: 'string', description: - 'Optional workflow ID to run. If not provided, uses the current workflow in context.', + 'ID of the workflow to run. Always pass this explicitly — outside a workflow-scoped chat there is no current workflow to fall back to, and the run is rejected without it.', }, workflow_input: { type: 'object', description: 'JSON object with key-value mappings where each key is an input field name', }, }, - required: ['startBlockId'], + required: ['workflowId', 'startBlockId'], }, clientExecutable: true, } @@ -4444,7 +4444,7 @@ export const RunWorkflow: ToolCatalogEntry = { workflowId: { type: 'string', description: - 'Optional workflow ID to run. If not provided, uses the current workflow in context.', + 'ID of the workflow to run. Always pass this explicitly — outside a workflow-scoped chat there is no current workflow to fall back to, and the run is rejected without it.', }, workflow_input: { type: 'object', @@ -4452,6 +4452,7 @@ export const RunWorkflow: ToolCatalogEntry = { "JSON object matching the target trigger's inputSchema (from get_workflow_run_options). For external/webhook triggers this is the event payload; for API/Input triggers it is the form fields.", }, }, + required: ['workflowId'], }, clientExecutable: true, requiresApproval: true, @@ -4492,7 +4493,7 @@ export const RunWorkflowUntilBlock: ToolCatalogEntry = { workflowId: { type: 'string', description: - 'Optional workflow ID to run. If not provided, uses the current workflow in context.', + 'ID of the workflow to run. Always pass this explicitly — outside a workflow-scoped chat there is no current workflow to fall back to, and the run is rejected without it.', }, workflow_input: { type: 'object', @@ -4500,7 +4501,7 @@ export const RunWorkflowUntilBlock: ToolCatalogEntry = { "JSON object matching the target trigger's inputSchema (from get_workflow_run_options). For external/webhook triggers this is the event payload; for API/Input triggers it is the form fields.", }, }, - required: ['stopAfterBlockId'], + required: ['workflowId', 'stopAfterBlockId'], }, clientExecutable: true, requiresApproval: true, diff --git a/apps/sim/lib/copilot/generated/tool-schemas-v1.ts b/apps/sim/lib/copilot/generated/tool-schemas-v1.ts index a7d92662bf..ccc7a598c0 100644 --- a/apps/sim/lib/copilot/generated/tool-schemas-v1.ts +++ b/apps/sim/lib/copilot/generated/tool-schemas-v1.ts @@ -4145,14 +4145,14 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { workflowId: { type: 'string', description: - 'Optional workflow ID to run. If not provided, uses the current workflow in context.', + 'ID of the workflow to run. Always pass this explicitly — outside a workflow-scoped chat there is no current workflow to fall back to, and the run is rejected without it.', }, workflow_input: { type: 'object', description: 'JSON object with key-value mappings where each key is an input field name', }, }, - required: ['blockId'], + required: ['workflowId', 'blockId'], }, resultSchema: undefined, }, @@ -4270,14 +4270,14 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { workflowId: { type: 'string', description: - 'Optional workflow ID to run. If not provided, uses the current workflow in context.', + 'ID of the workflow to run. Always pass this explicitly — outside a workflow-scoped chat there is no current workflow to fall back to, and the run is rejected without it.', }, workflow_input: { type: 'object', description: 'JSON object with key-value mappings where each key is an input field name', }, }, - required: ['startBlockId'], + required: ['workflowId', 'startBlockId'], }, resultSchema: undefined, }, @@ -4313,7 +4313,7 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { workflowId: { type: 'string', description: - 'Optional workflow ID to run. If not provided, uses the current workflow in context.', + 'ID of the workflow to run. Always pass this explicitly — outside a workflow-scoped chat there is no current workflow to fall back to, and the run is rejected without it.', }, workflow_input: { type: 'object', @@ -4321,6 +4321,7 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { "JSON object matching the target trigger's inputSchema (from get_workflow_run_options). For external/webhook triggers this is the event payload; for API/Input triggers it is the form fields.", }, }, + required: ['workflowId'], }, resultSchema: undefined, }, @@ -4355,7 +4356,7 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { workflowId: { type: 'string', description: - 'Optional workflow ID to run. If not provided, uses the current workflow in context.', + 'ID of the workflow to run. Always pass this explicitly — outside a workflow-scoped chat there is no current workflow to fall back to, and the run is rejected without it.', }, workflow_input: { type: 'object', @@ -4363,7 +4364,7 @@ export const TOOL_RUNTIME_SCHEMAS: Record = { "JSON object matching the target trigger's inputSchema (from get_workflow_run_options). For external/webhook triggers this is the event payload; for API/Input triggers it is the form fields.", }, }, - required: ['stopAfterBlockId'], + required: ['workflowId', 'stopAfterBlockId'], }, resultSchema: undefined, }, diff --git a/apps/sim/lib/copilot/generated/trace-attribute-values-v1.ts b/apps/sim/lib/copilot/generated/trace-attribute-values-v1.ts index 1fad4c1132..916ecc8856 100644 --- a/apps/sim/lib/copilot/generated/trace-attribute-values-v1.ts +++ b/apps/sim/lib/copilot/generated/trace-attribute-values-v1.ts @@ -117,6 +117,16 @@ export const CopilotConfirmOutcome = { export type CopilotConfirmOutcomeKey = keyof typeof CopilotConfirmOutcome export type CopilotConfirmOutcomeValue = (typeof CopilotConfirmOutcome)[CopilotConfirmOutcomeKey] +export const CopilotDegradedReason = { + BindingRejected: 'binding_rejected', + ClientPickupTimeout: 'client_pickup_timeout', + MissingToolResult: 'missing_tool_result', + StreamDeadBeforeDispatch: 'stream_dead_before_dispatch', +} as const + +export type CopilotDegradedReasonKey = keyof typeof CopilotDegradedReason +export type CopilotDegradedReasonValue = (typeof CopilotDegradedReason)[CopilotDegradedReasonKey] + export const CopilotFinalizeOutcome = { Aborted: 'aborted', Error: 'error', diff --git a/apps/sim/lib/copilot/generated/trace-attributes-v1.ts b/apps/sim/lib/copilot/generated/trace-attributes-v1.ts index 6db17a6329..5a026a33c3 100644 --- a/apps/sim/lib/copilot/generated/trace-attributes-v1.ts +++ b/apps/sim/lib/copilot/generated/trace-attributes-v1.ts @@ -189,6 +189,7 @@ export const TraceAttr = { CopilotCommandsCount: 'copilot.commands.count', CopilotConfirmOutcome: 'copilot.confirm.outcome', CopilotContextsCount: 'copilot.contexts.count', + CopilotDegradedReason: 'copilot.degraded.reason', CopilotExecutionId: 'copilot.execution.id', CopilotFileAttachmentsCount: 'copilot.file_attachments.count', CopilotFinalizeOutcome: 'copilot.finalize.outcome', @@ -833,6 +834,7 @@ export const TraceAttrValues: readonly TraceAttrValue[] = [ 'copilot.commands.count', 'copilot.confirm.outcome', 'copilot.contexts.count', + 'copilot.degraded.reason', 'copilot.execution.id', 'copilot.file_attachments.count', 'copilot.finalize.outcome', diff --git a/apps/sim/lib/copilot/request/handlers/handlers.test.ts b/apps/sim/lib/copilot/request/handlers/handlers.test.ts index a6ace4a3ac..32562b8293 100644 --- a/apps/sim/lib/copilot/request/handlers/handlers.test.ts +++ b/apps/sim/lib/copilot/request/handlers/handlers.test.ts @@ -143,6 +143,57 @@ describe('sse-handlers tool lifecycle', () => { } }) + it('pins the workflow target into the args it persists and forwards', async () => { + // The browser resolved its own target from the open tab while the server + // resolved the run's workflow; in a workspace chat those disagreed and every + // omitted-argument call was rejected. One stamped field ends that. + isSimExecuted.mockReturnValue(false) + context.runId = 'run-1' + const event = { + type: MothershipStreamV1EventType.tool, + payload: { + toolCallId: 'run-workflow-1', + toolName: 'run_workflow', + arguments: {}, + executor: MothershipStreamV1ToolExecutor.client, + mode: MothershipStreamV1ToolMode.async, + phase: MothershipStreamV1ToolPhase.call, + }, + } satisfies StreamEvent + + await prePersistClientExecutableToolCall(event, context, {}, execContext) + + // Forwarded frame — this is what the browser POSTs back with. + expect((event.payload as { arguments?: Record }).arguments).toEqual({ + workflowId: 'workflow-1', + }) + expect(upsertAsyncToolCall).toHaveBeenCalledWith( + expect.objectContaining({ toolCallId: 'run-workflow-1', args: { workflowId: 'workflow-1' } }) + ) + }) + + it('leaves an explicit workflow target untouched', async () => { + isSimExecuted.mockReturnValue(false) + context.runId = 'run-1' + const event = { + type: MothershipStreamV1EventType.tool, + payload: { + toolCallId: 'run-workflow-2', + toolName: 'run_workflow', + arguments: { workflowId: 'workflow-explicit' }, + executor: MothershipStreamV1ToolExecutor.client, + mode: MothershipStreamV1ToolMode.async, + phase: MothershipStreamV1ToolPhase.call, + }, + } satisfies StreamEvent + + await prePersistClientExecutableToolCall(event, context, {}, execContext) + + expect(upsertAsyncToolCall).toHaveBeenCalledWith( + expect.objectContaining({ args: { workflowId: 'workflow-explicit' } }) + ) + }) + it('pre-persists browser tools as pending for the desktop authorization claim', async () => { isSimExecuted.mockReturnValue(false) context.runId = 'run-1' @@ -735,6 +786,53 @@ describe('sse-handlers tool lifecycle', () => { expect(executeTool.mock.calls.at(-1)?.[2]?.workflowId).toBe('workflow-1') }) + it('refuses a workflow tool call with no resolvable workflow target', async () => { + // A workspace chat has no run-scoped workflow, so an omitted workflowId + // cannot be resolved by anyone on this side. Dispatching would only buy a + // rejection the model cannot read, so fail with something it can act on. + const workspaceExecContext = { ...execContext, workflowId: '' } + const onEvent = vi.fn() + + await sseHandlers.tool( + { + type: MothershipStreamV1EventType.tool, + payload: { + toolCallId: 'tool-unbound-workflow', + toolName: 'run_workflow', + arguments: {}, + executor: MothershipStreamV1ToolExecutor.client, + mode: MothershipStreamV1ToolMode.async, + phase: MothershipStreamV1ToolPhase.call, + }, + } satisfies StreamEvent, + context, + workspaceExecContext, + { onEvent, interactive: true, timeout: 1000 } + ) + + await Promise.allSettled(context.pendingToolPromises.values()) + + // Never handed to a browser, and never claimed server-side. + expect(waitForWorkflowToolCompletion).not.toHaveBeenCalled() + expect(claimWorkflowToolExecution).not.toHaveBeenCalled() + + const results = onEvent.mock.calls + .map(([event]) => event) + .filter( + (event) => + event?.type === MothershipStreamV1EventType.tool && + event.payload?.toolCallId === 'tool-unbound-workflow' && + event.payload?.phase === MothershipStreamV1ToolPhase.result + ) + expect(results).toHaveLength(1) + expect(results[0].payload.status).toBe(MothershipStreamV1ToolOutcome.error) + // The message has to name the fix, or the model just retries identically. + expect(results[0].payload.output?.error).toContain('workflowId') + expect(context.toolCalls.get('tool-unbound-workflow')?.status).toBe( + MothershipStreamV1ToolOutcome.error + ) + }) + it('does not run a workflow tool server-side when a browser holds the claim', async () => { waitForWorkflowToolCompletion.mockResolvedValue(null) claimWorkflowToolExecution.mockResolvedValueOnce(null) @@ -1488,6 +1586,40 @@ describe('sse-handlers tool lifecycle', () => { expect(context.pendingToolPromises.has('tool-inflight')).toBe(false) }) + it('leaves a complete terminal state when a tool is cancelled before dispatch', async () => { + // A tool cancelled because its stream was already aborted used to get a + // status but no `result`. The subagent join requires one, so that single + // half-finished tool call was turned into a thrown "missing result" that + // killed the entire turn and blamed an unrelated tool. + context.wasAborted = true + + await sseHandlers.tool( + { + type: MothershipStreamV1EventType.tool, + payload: { + toolCallId: 'tool-stream-dead', + toolName: ReadTool.id, + arguments: { workflowId: 'workflow-1' }, + executor: MothershipStreamV1ToolExecutor.sim, + mode: MothershipStreamV1ToolMode.async, + phase: MothershipStreamV1ToolPhase.call, + }, + } satisfies StreamEvent, + context, + execContext, + { onEvent: vi.fn(), interactive: false, timeout: 1000 } + ) + + await sleep(0) + + const toolCall = context.toolCalls.get('tool-stream-dead') + expect(executeTool).not.toHaveBeenCalled() + expect(toolCall?.status).toBe(MothershipStreamV1ToolOutcome.cancelled) + // The part that was missing: a terminal tool must also be complete. + expect(toolCall?.result).toEqual({ success: false }) + expect(toolCall?.error).toBeTruthy() + }) + it('still executes the tool when async row upsert fails', async () => { upsertAsyncToolCall.mockRejectedValueOnce(new Error('db down')) executeTool.mockResolvedValueOnce({ success: true, output: { ok: true } }) diff --git a/apps/sim/lib/copilot/request/handlers/tool.ts b/apps/sim/lib/copilot/request/handlers/tool.ts index d08f04b486..c7e340b013 100644 --- a/apps/sim/lib/copilot/request/handlers/tool.ts +++ b/apps/sim/lib/copilot/request/handlers/tool.ts @@ -226,6 +226,22 @@ export async function prePersistClientExecutableToolCall( if (!context.runId) return + // Pin the workflow target into the arguments before they are sealed, persisted, + // and forwarded, so the row, the browser, and the completion waiter all read one + // explicit field. + // + // They used to disagree: the server resolved `args.workflowId ?? run.workflowId` + // while the browser resolved `args.workflowId ?? activeWorkflowId`. In a + // workspace chat `copilot_runs.workflow_id` is NULL, so every call that omitted + // the (optional) argument resolved to nothing server-side and to the open tab + // client-side — a guaranteed rejection at the execute endpoint. + if (isWorkflowToolName(data.toolName)) { + const targetWorkflowId = resolveWorkflowToolTargetId(data.arguments, execContext?.workflowId) + if (targetWorkflowId) { + data.arguments = { ...(data.arguments ?? {}), workflowId: targetWorkflowId } + } + } + let sealedContext: Awaited> | undefined if (execContext?.resolvedSecretTraceRegistry) { try { @@ -691,6 +707,39 @@ async function dispatchToolExecution( }) } + /** + * Refuse a workflow tool call whose target this side cannot name. + * + * The execute endpoint validates the run against the tool call's bound + * workflow, so dispatching an unbound call only buys a rejection the model + * cannot interpret. Failing here instead tells it exactly what to send back, + * and never guesses a workflow on the user's behalf. + */ + const refuseUnboundWorkflowTool = async (): Promise => { + const error = `${toolName} requires an explicit workflowId. This chat is not scoped to a workflow, so there is no current workflow to fall back to — pass the id of the workflow to run.` + logger.warn('Refusing workflow tool call with no resolvable workflow target', { + toolCallId, + toolName, + }) + setTerminalToolCallState(toolCall, { + status: MothershipStreamV1ToolOutcome.error, + output: { error }, + error, + }) + markToolResultSeen(toolCallId) + await emitSyntheticToolResult( + toolCallId, + toolCall.name, + { + status: MothershipStreamV1ToolOutcome.error, + message: error, + data: { error }, + }, + options + ) + return { status: MothershipStreamV1ToolOutcome.error, message: error, data: { error } } + } + // Returns the promise instead of registering it, so the permission gate can // wrap the whole thing in one pending promise that stays unsettled until the // tool has actually run. Null means nothing was dispatched. @@ -708,6 +757,12 @@ async function dispatchToolExecution( if (abortPendingToolIfStreamDead(toolCall, toolCallId, options, context)) return null return fireToolExecution() } + if ( + delegateWorkflowRunToClient && + !resolveWorkflowToolTargetId(args, execContext.workflowId) + ) { + return refuseUnboundWorkflowTool() + } return waitForClientExecution() } diff --git a/apps/sim/lib/copilot/request/handlers/types.ts b/apps/sim/lib/copilot/request/handlers/types.ts index 3e383e2c6f..a7f9d81946 100644 --- a/apps/sim/lib/copilot/request/handlers/types.ts +++ b/apps/sim/lib/copilot/request/handlers/types.ts @@ -16,6 +16,8 @@ import { MothershipStreamV1ToolPhase, type MothershipStreamV1ToolResultPayload, } from '@/lib/copilot/generated/mothership-stream-v1' +import { CopilotDegradedReason } from '@/lib/copilot/generated/trace-attribute-values-v1' +import { recordDegraded } from '@/lib/copilot/request/metrics' import { asRecord, markToolResultSeen } from '@/lib/copilot/request/sse-utils' import { setTerminalToolCallState } from '@/lib/copilot/request/tool-call-state' import type { @@ -149,17 +151,37 @@ export function abortPendingToolIfStreamDead( if (!options.abortSignal?.aborted && !context.wasAborted) { return false } - toolCall.status = MothershipStreamV1ToolOutcome.cancelled - toolCall.endTime = Date.now() + const abortReason = options.abortSignal?.aborted + ? String(options.abortSignal.reason ?? 'unknown') + : undefined + // Go through the canonical terminal helper rather than stamping status by + // hand: it also writes `result`, and everything downstream that reads a + // finished tool call requires one. Leaving it unset made this call look + // terminal-but-incomplete, which the subagent join turned into a thrown + // "missing result" error that killed the whole turn. + setTerminalToolCallState(toolCall, { + status: MothershipStreamV1ToolOutcome.cancelled, + error: 'Tool was not dispatched because its stream had already been aborted', + }) markToolResultSeen(toolCallId) + // Sim's logs do not reach Loki and the trace span below is collected + // in-process but never exported, so the counter is the only signal that + // survives to somewhere queryable. + recordDegraded(CopilotDegradedReason.StreamDeadBeforeDispatch) + logger.warn('Cancelled tool call before dispatch: stream already aborted', { + toolCallId, + toolName: toolCall.name, + reason: 'stream_dead_before_dispatch', + abortSignalAborted: options.abortSignal?.aborted ?? false, + ...(abortReason ? { abortReason } : {}), + wasAborted: context.wasAborted ?? false, + }) const toolSpan = context.trace.startSpan(toolCall.name || 'unknown_tool', 'tool.execute', { toolCallId, toolName: toolCall.name, cancelReason: 'stream_dead_before_dispatch', abortSignalAborted: options.abortSignal?.aborted ?? false, - abortReason: options.abortSignal?.aborted - ? String(options.abortSignal.reason ?? 'unknown') - : undefined, + abortReason, wasAborted: context.wasAborted ?? false, }) context.trace.endSpan(toolSpan, 'cancelled') diff --git a/apps/sim/lib/copilot/request/lifecycle/resume-leg-context.test.ts b/apps/sim/lib/copilot/request/lifecycle/resume-leg-context.test.ts index 68fb457f07..37fff6f90c 100644 --- a/apps/sim/lib/copilot/request/lifecycle/resume-leg-context.test.ts +++ b/apps/sim/lib/copilot/request/lifecycle/resume-leg-context.test.ts @@ -9,6 +9,10 @@ import { makeResumeLegContext, mergeResumeLegOutputs } from '@/lib/copilot/reque // all concurrent legs build one chat. This is the regression the inline comment // warns about — without per-leg isolation the orchestrator's pre-fanout content // gets multiplied by the leg count on merge. +// +// `wasAborted` is the one deliberate exception to "reset AND folded back": it is +// reset per leg but folded back only for a turn-level abort, because a fanout +// cancelling its own lanes must not mark the shared turn aborted. describe('resume leg context isolate/merge contract', () => { it('isolates the per-leg scalars while sharing the heavy accumulators by reference', () => { const base = createStreamingContext({ @@ -31,6 +35,9 @@ describe('resume leg context isolate/merge contract', () => { expect(leg.streamComplete).toBe(false) expect(leg.awaitingAsyncContinuation).toBeUndefined() expect(leg.completionStatus).toBeUndefined() + // A leg must never be born aborted — that is what let one cancelled lane + // cancel every tool dispatched on every lane created after it. + expect(leg.wasAborted).toBe(false) // A leg's own errors array is a fresh array (not the shared one) so a leg's // retry rollback can't truncate a sibling's errors. @@ -65,6 +72,34 @@ describe('resume leg context isolate/merge contract', () => { expect(base.completionStatus).toBe(MothershipStreamV1CompletionStatus.complete) }) + it('does not fold a fanout-induced abort onto the shared turn', () => { + // A lane that fails cancels its siblings by design, and each cancelled + // sibling returns normally with wasAborted set. Folding that marked the + // SHARED context aborted, so every leg created afterwards was born aborted + // and every tool it dispatched was cancelled before dispatch — which the + // subagent join then reported as a fatal "missing result". + const base = createStreamingContext({}) + const leg = makeResumeLegContext(base) + leg.wasAborted = true + + mergeResumeLegOutputs(base, leg, false) + + expect(base.wasAborted).toBe(false) + }) + + it('folds a turn-level abort onto the shared turn', () => { + // The other half: a real Stop (or an observed abort marker) must reach the + // shared context, because that is what classifies the request as cancelled + // rather than successful. + const base = createStreamingContext({}) + const leg = makeResumeLegContext(base) + leg.wasAborted = true + + mergeResumeLegOutputs(base, leg, true) + + expect(base.wasAborted).toBe(true) + }) + it('leaves the turn unfinished when only child legs fold back', () => { const base = createStreamingContext() diff --git a/apps/sim/lib/copilot/request/lifecycle/run.test.ts b/apps/sim/lib/copilot/request/lifecycle/run.test.ts index 5f535532f5..4b86ccb104 100644 --- a/apps/sim/lib/copilot/request/lifecycle/run.test.ts +++ b/apps/sim/lib/copilot/request/lifecycle/run.test.ts @@ -1759,6 +1759,61 @@ describe('runCopilotLifecycle', () => { } }) + it('retries an initial stream that ended before its checkpoint pause with one request identity', async () => { + const headers: Array> = [] + const executionContext: ExecutionContext = { + userId: 'user-1', + workflowId: '', + workspaceId: 'ws-1', + chatId: 'chat-1', + } + + mockRunStreamLoop.mockImplementationOnce( + async ( + _fetchUrl: string, + fetchOptions: RequestInit, + context: StreamingContext + ): Promise => { + headers.push(fetchOptions.headers as Record) + context.errors.push(STREAM_ENDED_WITHOUT_TERMINAL_MESSAGE) + throw new StreamEndedWithoutTerminalError('/api/mothership') + } + ) + mockRunStreamLoop.mockImplementationOnce( + async ( + _fetchUrl: string, + fetchOptions: RequestInit, + context: StreamingContext + ): Promise => { + headers.push(fetchOptions.headers as Record) + context.streamComplete = true + context.completionStatus = MothershipStreamV1CompletionStatus.complete + } + ) + + const result = await runCopilotLifecycle( + { message: 'hello', messageId: 'stream-initial-retry' }, + { + userId: 'user-1', + workspaceId: 'ws-1', + chatId: 'chat-1', + executionId: 'exec-1', + runId: 'run-1', + simRequestId: 'request-initial-retry', + executionContext, + } + ) + + expect(mockRunStreamLoop).toHaveBeenCalledTimes(2) + expect(headers.map((value) => value['X-Sim-Request-ID'])).toEqual([ + 'request-initial-retry', + 'request-initial-retry', + ]) + expect(result).toEqual( + expect.objectContaining({ success: true, cancelled: false, errors: undefined }) + ) + }) + it('does not retry a resume leg the backend already claimed and ended early', async () => { const executionContext: ExecutionContext = { userId: 'user-1', @@ -2121,4 +2176,103 @@ describe('runCopilotLifecycle', () => { vi.useRealTimers() } }) + it('completes the turn when a pending subagent tool has no result', async () => { + // A tool that never reached a terminal state used to throw + // "Cannot resume subagent chain ...: missing result for tool call ...", + // which inside a fanout cancelled every sibling lane and reported the whole + // request as an error blaming an unrelated tool. Synthesize the failure Go + // already writes for itself instead, and let the turn finish. + const bodies: Array> = [] + mockRunStreamLoop.mockImplementation( + async ( + fetchUrl: string, + fetchOptions: RequestInit, + context: StreamingContext + ): Promise => { + if (!fetchUrl.includes('/api/tools/resume')) { + context.awaitingAsyncContinuation = { + checkpointId: 'cp-root', + pendingToolCallIds: [], + frames: [ + { + parentToolCallId: 'subagent-file', + parentToolName: 'file', + pendingToolIds: ['tool-never-dispatched'], + checkpointId: 'cp-file', + }, + ], + } + return + } + bodies.push(JSON.parse(String(fetchOptions.body))) + context.streamComplete = true + context.completionStatus = MothershipStreamV1CompletionStatus.complete + } + ) + + const result = await runCopilotLifecycle( + { message: 'hello', messageId: 'stream-missing-subagent-result' }, + { userId: 'user-1', workspaceId: 'ws-1' } + ) + + expect(bodies).toHaveLength(1) + expect(bodies[0].checkpointId).toBe('cp-file') + expect(bodies[0].results).toEqual([ + expect.objectContaining({ + callId: 'tool-never-dispatched', + success: false, + data: { error: expect.stringContaining('no result was returned') }, + }), + ]) + expect(result.success).toBe(true) + }) + + it('classifies a Stop landing during a subagent fanout as cancelled', async () => { + // Guards the trap in the fanout fix: `wasAborted` is now isolated per leg, so + // a user Stop must still reach the turn — via the abort signal or the folded + // turn-level abort — or a cancelled turn would be reported as a success. + const controller = new AbortController() + mockRunStreamLoop.mockImplementation( + async ( + fetchUrl: string, + _fetchOptions: RequestInit, + context: StreamingContext + ): Promise => { + if (!fetchUrl.includes('/api/tools/resume')) { + context.toolCalls.set('tool-done', { + id: 'tool-done', + name: 'read', + status: MothershipStreamV1ToolOutcome.success, + result: { success: true }, + endTime: Date.now(), + }) + context.awaitingAsyncContinuation = { + checkpointId: 'cp-root', + pendingToolCallIds: [], + frames: [ + { + parentToolCallId: 'subagent-file', + parentToolName: 'file', + pendingToolIds: ['tool-done'], + checkpointId: 'cp-file', + }, + ], + } + return + } + // The user hits Stop mid-fanout. `wasAborted` is isolated per leg now, so + // the turn's own signal is what has to carry the cancellation into the + // classification — reading only `context.wasAborted` reports success. + controller.abort('user_stop') + } + ) + + const result = await runCopilotLifecycle( + { message: 'hello', messageId: 'stream-stop-during-fanout' }, + { userId: 'user-1', workspaceId: 'ws-1', abortSignal: controller.signal } + ) + + expect(result.cancelled).toBe(true) + expect(result.success).toBe(false) + }) }) diff --git a/apps/sim/lib/copilot/request/lifecycle/run.ts b/apps/sim/lib/copilot/request/lifecycle/run.ts index 389f31564a..f828866199 100644 --- a/apps/sim/lib/copilot/request/lifecycle/run.ts +++ b/apps/sim/lib/copilot/request/lifecycle/run.ts @@ -28,6 +28,7 @@ import { MothershipStreamV1RunKind, MothershipStreamV1ToolOutcome, } from '@/lib/copilot/generated/mothership-stream-v1' +import { CopilotDegradedReason } from '@/lib/copilot/generated/trace-attribute-values-v1' import { getAutoAllowedTools } from '@/lib/copilot/persistence/tool-permission/auto-allow' import { createStreamingContext } from '@/lib/copilot/request/context/request-context' import { buildToolCallSummaries } from '@/lib/copilot/request/context/result' @@ -37,6 +38,8 @@ import { runStreamLoop, StreamEndedWithoutTerminalError, } from '@/lib/copilot/request/go/stream' +import { recordDegraded } from '@/lib/copilot/request/metrics' +import { AbortReason } from '@/lib/copilot/request/session/abort-reason' import { getToolCallTerminalData, requireToolCallStateResult, @@ -346,7 +349,13 @@ export async function runCopilotLifecycle( // the work the user watched succeed. const backendFinishedTurn = context.completionStatus === MothershipStreamV1CompletionStatus.complete - const succeeded = !context.wasAborted && (backendFinishedTurn || context.errors.length === 0) + // Consult the lifecycle signal as well as the flag. `context.wasAborted` is + // only reached from a fanout leg through the (deliberately asymmetric) merge + // in `mergeResumeLegOutputs`, so a Stop landing mid-fanout could otherwise + // classify the turn as a success. Mirrors the check already used below on + // the throw path. + const turnWasAborted = context.wasAborted || (lifecycleOptions.abortSignal?.aborted ?? false) + const succeeded = !turnWasAborted && (backendFinishedTurn || context.errors.length === 0) const result: OrchestratorResult = { success: succeeded, @@ -359,7 +368,7 @@ export async function runCopilotLifecycle( // path, but practically that doesn't happen in the success // branch here — if there are errors we never reach a // wasAborted-without-errors state. - cancelled: context.wasAborted && context.errors.length === 0, + cancelled: turnWasAborted && context.errors.length === 0, content: resultContent(context, lifecycleOptions), contentBlocks: context.contentBlocks, toolCalls: buildToolCallSummaries(context), @@ -463,10 +472,12 @@ function isPerSubagentContinuation(c: AsyncContinuation): boolean { // every resume leg), so the auth/source/version headers can't drift between the // sequential path and the concurrent per-subagent resume legs. function mothershipRequestHeaders( - hostedBillingRequest?: AttributedBillingRequestEnvelope + hostedBillingRequest?: AttributedBillingRequestEnvelope, + simRequestId?: string ): Record { return { 'Content-Type': 'application/json', + ...(simRequestId ? { 'X-Sim-Request-ID': simRequestId } : {}), ...(env.COPILOT_API_KEY ? { 'x-api-key': env.COPILOT_API_KEY } : {}), ...getMothershipSourceEnvHeaders(), 'X-Client-Version': SIM_AGENT_VERSION, @@ -498,6 +509,12 @@ function mothershipRequestHeaders( // - completionStatus: the backend's terminal verdict, set only on the leg that // carries the turn to its end; a stale one from a sibling would speak for a // turn that leg never finished. +// - wasAborted: the ONE field with an asymmetric fold. Cancelling a fanout +// cancels its siblings by design, and each cancelled sibling returns +// normally with wasAborted set — folding that unconditionally marked the +// SHARED context aborted, so every later leg was born aborted and every tool +// it dispatched was cancelled before dispatch. Reset per leg, and fold back +// only for a turn-level abort (see mergeResumeLegOutputs). // When adding a per-leg field, update BOTH functions (and the contract test in // resume-leg-context.test.ts). Exported only for that test. export function makeResumeLegContext(base: StreamingContext): StreamingContext { @@ -511,19 +528,30 @@ export function makeResumeLegContext(base: StreamingContext): StreamingContext { cost: undefined, errors: [], completionStatus: undefined, + wasAborted: false, } } // mergeResumeLegOutputs folds a finished leg's isolated scalars back into the // shared context. Child (subagent-lane) legs leave the join scalars empty; only // the join-carrying leg (which streams the orchestrator continuation) sets them. -export function mergeResumeLegOutputs(context: StreamingContext, leg: StreamingContext): void { +// +// `turnWasAborted` is the caller's answer to "was this abort the turn's, or just +// this fanout cancelling its own lanes?". Only a turn-level abort belongs on the +// shared context: it is what `runCopilotLifecycle` reads to classify the request +// as cancelled, and on the headless path (which never wires `onAbortObserved`) +// it is the only record that the abort marker was ever observed. +export function mergeResumeLegOutputs( + context: StreamingContext, + leg: StreamingContext, + turnWasAborted = true +): void { if (leg.accumulatedContent) context.accumulatedContent += leg.accumulatedContent if (leg.finalAssistantContent) context.finalAssistantContent += leg.finalAssistantContent if (leg.usage) context.usage = leg.usage if (leg.cost) context.cost = leg.cost if (leg.sawMainToolCall) context.sawMainToolCall = true - if (leg.wasAborted) context.wasAborted = true + if (leg.wasAborted && turnWasAborted) context.wasAborted = true if (leg.errors.length > 0) context.errors.push(...leg.errors) if (leg.completionStatus) context.completionStatus = leg.completionStatus } @@ -537,26 +565,61 @@ async function waitForToolIds(context: StreamingContext, toolIds: string[]): Pro if (promises.length > 0) await Promise.allSettled(promises) } +interface ResumeToolResult { + callId: string + name: string + data: unknown + success: boolean +} + +/** + * Build the resume payload entry for one pending tool call. + * + * A tool that never reached a terminal state has no result to send. That used to + * throw — which turned one incomplete tool into a dead turn, and inside a + * subagent fanout cancelled every sibling lane and reported the whole request as + * an error blaming an unrelated tool. Synthesize the same failure Go already + * writes for itself when Sim posts nothing for a pending call, so the model sees + * one failed tool and the turn carries on. Still logged at error level: getting + * here is a bug, it just must not be fatal. + */ +function buildResumeToolResult( + context: StreamingContext, + toolCallId: string, + checkpointId: string | undefined +): ResumeToolResult { + const tool = context.toolCalls.get(toolCallId) + if (!tool || !tool.result) { + recordDegraded(CopilotDegradedReason.MissingToolResult) + logger.error('Missing tool result for pending tool call; synthesizing a failure', { + toolCallId, + checkpointId, + hasToolEntry: !!tool, + toolName: tool?.name, + toolStatus: tool?.status, + hasPendingPromise: context.pendingToolPromises.has(toolCallId), + }) + return { + callId: toolCallId, + name: tool?.name || '', + data: { error: `no result was returned for tool call ${toolCallId}` }, + success: false, + } + } + return { + callId: toolCallId, + name: tool.name || '', + data: getToolCallTerminalData(tool), + success: requireToolCallStateResult(tool).success, + } +} + function collectResultsForToolIds( context: StreamingContext, toolIds: string[], checkpointId: string -): Array<{ callId: string; name: string; data: unknown; success: boolean }> { - return toolIds.map((toolCallId) => { - const tool = context.toolCalls.get(toolCallId) - if (!tool || !tool.result) { - throw new Error( - `Cannot resume subagent chain ${checkpointId}: missing result for tool call ${toolCallId}` - ) - } - const name = tool.name || '' - return { - callId: toolCallId, - name, - data: getToolCallTerminalData(tool), - success: requireToolCallStateResult(tool).success, - } - }) +): ResumeToolResult[] { + return toolIds.map((toolCallId) => buildResumeToolResult(context, toolCallId, checkpointId)) } // runResumeLegWithRetry runs ONE resume POST with the same retryable-error + @@ -583,7 +646,7 @@ async function runResumeLegWithRetry( url, { method: 'POST', - headers: mothershipRequestHeaders(hostedBillingRequest), + headers: mothershipRequestHeaders(hostedBillingRequest, options.simRequestId), body: JSON.stringify(body), }, leg, @@ -622,6 +685,12 @@ async function driveOneChildChain( execContext: ExecutionContext, options: CopilotLifecycleOptions, baseURL: string, + /** + * The turn's own abort signal, NOT the fanout controller in `options`. Used to + * tell "the user stopped the turn" from "a lane failed and cancelled its + * siblings" when deciding whether a leg's abort belongs on the shared context. + */ + turnAbortSignal: AbortSignal | undefined, workspaceId?: string, hostedBillingRequest?: AttributedBillingRequestEnvelope ): Promise { @@ -642,6 +711,18 @@ async function driveOneChildChain( const results = collectResultsForToolIds(context, toolIds, checkpointId) const leg = makeResumeLegContext(context) + // The abort marker is turn-scoped (keyed on the shared messageId), so a leg + // that observes it at body close IS a turn-level abort — and on the headless + // path, where `onAbortObserved` is never wired to the turn controller, this + // is the only record of it. + let markerObserved = false + const legOptions: CopilotLifecycleOptions = { + ...options, + onAbortObserved: (reason) => { + if (reason === AbortReason.MarkerObservedAtBodyClose) markerObserved = true + options.onAbortObserved?.(reason) + }, + } await runResumeLegWithRetry( `${baseURL}/api/tools/resume`, { @@ -653,10 +734,10 @@ async function driveOneChildChain( }, leg, execContext, - options, + legOptions, hostedBillingRequest ) - mergeResumeLegOutputs(context, leg) + mergeResumeLegOutputs(context, leg, markerObserved || (turnAbortSignal?.aborted ?? false)) const cont = leg.awaitingAsyncContinuation if (!cont) { @@ -735,6 +816,7 @@ async function driveSubagentChains( execContext, legOptions, baseURL, + parentSignal, workspaceId, hostedBillingRequest ).catch((error) => { @@ -769,9 +851,14 @@ async function runCheckpointLoop( let route = initialRoute let payload: Record = initialPayload let resumeAttempt = 0 + let initialAttempt = 0 const callerOnEvent = options.onEvent const mothershipBaseURL = await getMothershipBaseURL({ userId: options.userId }) const lifecycleWorkspaceId = nonBlankString(options.workspaceId) + const mothershipRequestId = nonBlankString(options.simRequestId) ?? generateId() + if (!options.simRequestId) { + options = { ...options, simRequestId: mothershipRequestId } + } const systemPromptOverride = env.MSHIP_SYSPROMPT_OVERRIDE if (typeof systemPromptOverride === 'string' && systemPromptOverride.trim() !== '') { @@ -855,7 +942,7 @@ async function runCheckpointLoop( `${mothershipBaseURL}${route}`, { method: 'POST', - headers: mothershipRequestHeaders(hostedBillingRequest), + headers: mothershipRequestHeaders(hostedBillingRequest, mothershipRequestId), body: JSON.stringify(payload), }, context, @@ -870,6 +957,7 @@ async function runCheckpointLoop( context.trace.endSpan(streamSpan, streamStatus) context.trace.setActiveSpan(undefined) resumeAttempt = 0 + initialAttempt = 0 } catch (streamError) { context.trace.endSpan(streamSpan, RequestTraceV1SpanStatus.error) context.trace.setActiveSpan(undefined) @@ -877,22 +965,27 @@ async function runCheckpointLoop( await handleBillingLimitResponse(streamError.userId, context, execContext, options) break } - if ( - isResume && - isRetryableStreamError(streamError) && - resumeAttempt < MAX_RESUME_ATTEMPTS - 1 - ) { + const attempt = isResume ? resumeAttempt : initialAttempt + const retryable = isResume + ? isRetryableStreamError(streamError) + : isRetryableInitialStreamError(streamError) + if (retryable && attempt < MAX_RESUME_ATTEMPTS - 1) { // Discard errors recorded during this failed attempt; we're about to // redo this leg and a clean retry must not finalize as `error`. context.errors.length = errorsBeforeAttempt - resumeAttempt++ - const backoff = RESUME_BACKOFF_MS[resumeAttempt - 1] ?? 1000 - logger.warn('Resume stream failed, retrying', { - attempt: resumeAttempt + 1, - maxAttempts: MAX_RESUME_ATTEMPTS, - backoffMs: backoff, - error: toError(streamError).message, - }) + if (isResume) resumeAttempt++ + else initialAttempt++ + const nextAttempt = isResume ? resumeAttempt : initialAttempt + const backoff = RESUME_BACKOFF_MS[nextAttempt - 1] ?? 1000 + logger.warn( + isResume ? 'Resume stream failed, retrying' : 'Initial stream failed, retrying', + { + attempt: nextAttempt + 1, + maxAttempts: MAX_RESUME_ATTEMPTS, + backoffMs: backoff, + error: toError(streamError).message, + } + ) await sleepWithAbort(backoff, options.abortSignal) continue } @@ -1047,37 +1140,14 @@ async function runCheckpointLoop( break } - const results: Array<{ - callId: string - name: string - data: unknown - success: boolean - }> = [] + const results: ResumeToolResult[] = [] for (const toolCallId of continuation.pendingToolCallIds) { if (isAborted(options, context)) { cancelPendingTools(context) context.awaitingAsyncContinuation = undefined break } - const tool = context.toolCalls.get(toolCallId) - if (!tool || !tool.result) { - logger.error('Missing tool result for pending tool call', { - toolCallId, - checkpointId: continuation.checkpointId, - hasToolEntry: !!tool, - toolName: tool?.name, - toolStatus: tool?.status, - hasPendingPromise: context.pendingToolPromises.has(toolCallId), - }) - throw new Error(`Cannot resume: missing result for pending tool call ${toolCallId}`) - } - const name = tool.name || '' - results.push({ - callId: toolCallId, - name, - data: getToolCallTerminalData(tool), - success: requireToolCallStateResult(tool).success, - }) + results.push(buildResumeToolResult(context, toolCallId, continuation.checkpointId)) } if (isAborted(options, context)) { @@ -1325,6 +1395,25 @@ function isRetryableStreamError(error: unknown): boolean { return false } +/** + * Initial requests use a durable request identity and the backend's checkpoint + * delivery reservation. Reposting a transport-ambiguous initial leg is safe: + * Go redelivers an untouched committed pause, starts a request that never + * anchored, or fails closed when the accepted leg has no recoverable pause. + */ +function isRetryableInitialStreamError(error: unknown): boolean { + if (error instanceof DOMException && error.name === 'AbortError') { + return false + } + if (error instanceof StreamEndedWithoutTerminalError) { + return true + } + if (error instanceof CopilotBackendError) { + return error.status !== undefined && error.status >= 500 + } + return error instanceof TypeError +} + function sleepWithAbort(ms: number, abortSignal?: AbortSignal): Promise { if (!abortSignal) { return sleep(ms) diff --git a/apps/sim/lib/copilot/request/metrics.ts b/apps/sim/lib/copilot/request/metrics.ts index d3bfb80438..7ad1ed582a 100644 --- a/apps/sim/lib/copilot/request/metrics.ts +++ b/apps/sim/lib/copilot/request/metrics.ts @@ -11,6 +11,7 @@ import { type Counter, type Histogram, metrics } from '@opentelemetry/api' import { Metric } from '@/lib/copilot/generated/metrics-v1' import { TOOL_CATALOG } from '@/lib/copilot/generated/tool-catalog-v1' +import type { CopilotDegradedReasonValue } from '@/lib/copilot/generated/trace-attribute-values-v1' import { TraceAttr } from '@/lib/copilot/generated/trace-attributes-v1' // MUST match Go's copilot/internal/telemetry/metrics.go LatencyBucketsMs @@ -26,6 +27,7 @@ const BYTE_BUCKETS = [1024, 8192, 65536, 262144, 1048576, 4194304, 16777216, 671 interface CopilotMeterInstruments { toolDuration: Histogram toolCalls: Counter + degradedCount: Counter vfsMaterializeDuration: Histogram fileReadDuration: Histogram fileReadBytes: Histogram @@ -45,6 +47,7 @@ function instruments(): CopilotMeterInstruments { advice: { explicitBucketBoundaries: LATENCY_BUCKETS_MS }, }), toolCalls: meter.createCounter(Metric.CopilotToolCalls), + degradedCount: meter.createCounter(Metric.CopilotDegradedCount), vfsMaterializeDuration: meter.createHistogram(Metric.CopilotVfsMaterializeDuration, { unit: 'ms', advice: { explicitBucketBoundaries: LATENCY_BUCKETS_MS }, @@ -97,6 +100,18 @@ export function recordSimToolMetric( if (durationMs >= 0) toolDuration.record(durationMs, attrs) } +// recordDegraded counts one non-fatal fallback, labelled by the bounded reason +// it took. Every degradation path reports here so "are we degrading, and why" is +// one query instead of a per-incident investigation — and so a path that should +// be impossible can be alerted on at > 0. Sim's own logs do not reach Loki and +// the in-process TraceCollector is not exported, so without this a fallback is +// invisible. +export function recordDegraded(reason: CopilotDegradedReasonValue): void { + instruments().degradedCount.add(1, { + [TraceAttr.CopilotDegradedReason]: reason, + }) +} + // recordVfsMaterialize records VFS materialization time. Call once per phase // with that phase's duration and once with phase="total" for the whole op, so // the dashboard can show total + per-phase. phase must be a bounded value. diff --git a/apps/sim/lib/copilot/request/tools/workflow-client-fallback.test.ts b/apps/sim/lib/copilot/request/tools/workflow-client-fallback.test.ts index 328d62c6d3..d786190a33 100644 --- a/apps/sim/lib/copilot/request/tools/workflow-client-fallback.test.ts +++ b/apps/sim/lib/copilot/request/tools/workflow-client-fallback.test.ts @@ -4,10 +4,15 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { waitForWorkflowToolCompletion, claimWorkflowToolExecution } = vi.hoisted(() => ({ - waitForWorkflowToolCompletion: vi.fn(), - claimWorkflowToolExecution: vi.fn(), -})) +const { waitForWorkflowToolCompletion, claimWorkflowToolExecution, recordDegraded } = vi.hoisted( + () => ({ + waitForWorkflowToolCompletion: vi.fn(), + claimWorkflowToolExecution: vi.fn(), + recordDegraded: vi.fn(), + }) +) + +vi.mock('@/lib/copilot/request/metrics', () => ({ recordDegraded })) vi.mock('@/lib/copilot/request/tools/client', () => ({ waitForWorkflowToolCompletion, @@ -84,6 +89,9 @@ describe('raceWorkflowToolClientPickup', () => { expect(outcome.winner).toBe('sim') expect(outcome.signal).toEqual({ status: 'success' }) + // Falling back is non-fatal, so it has to be countable — Sim logs do not + // reach Loki and this path emits no exported span. + expect(recordDegraded).toHaveBeenCalledWith('client_pickup_timeout') expect(claimWorkflowToolExecution).toHaveBeenCalledTimes(1) expect(params.runOnServer).toHaveBeenCalledTimes(1) // The claimed id is what the server run must bind to. diff --git a/apps/sim/lib/copilot/request/tools/workflow-client-fallback.ts b/apps/sim/lib/copilot/request/tools/workflow-client-fallback.ts index 5e63209ef6..6a5a2df4c0 100644 --- a/apps/sim/lib/copilot/request/tools/workflow-client-fallback.ts +++ b/apps/sim/lib/copilot/request/tools/workflow-client-fallback.ts @@ -7,6 +7,8 @@ import type { AsyncTerminalCompletionSnapshot, } from '@/lib/copilot/async-runs/lifecycle' import { claimWorkflowToolExecution } from '@/lib/copilot/async-runs/repository' +import { CopilotDegradedReason } from '@/lib/copilot/generated/trace-attribute-values-v1' +import { recordDegraded } from '@/lib/copilot/request/metrics' import { waitForWorkflowToolCompletion } from '@/lib/copilot/request/tools/client' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' @@ -121,6 +123,7 @@ export async function raceWorkflowToolClientPickup( return { winner: 'client', completion: await clientWait } } + recordDegraded(CopilotDegradedReason.ClientPickupTimeout) logger.info('No client picked up workflow tool within grace; running it server-side', { toolCallId, workflowId, diff --git a/apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts b/apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts index 7b4817d0b8..ccf8c55de2 100644 --- a/apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts +++ b/apps/sim/lib/copilot/tools/client/run-tool-execution.test.ts @@ -451,6 +451,38 @@ describe('run tool execution cancellation', () => { ) }) + it('reports the real failure reason so the agent can correct its arguments', async () => { + // A generic "Workflow execution failed." told the model nothing, so it could + // not fix a rejected binding or an undeployed workflow on retry. + const fetchMock = vi.fn().mockResolvedValue({ ok: true }) + vi.stubGlobal('fetch', fetchMock) + executeWorkflowWithFullLogging.mockRejectedValueOnce( + new MockExecutionStreamHttpError( + 'This Copilot workflow tool call is bound to a different workflow', + 403, + 'COPILOT_WORKFLOW_TOOL_BINDING_WORKFLOW_MISMATCH' + ) + ) + + executeRunToolOnClient('tool-binding-rejected', 'run_workflow', { workflowId: 'wf-1' }) + + await vi.waitFor(() => { + expect(fetchMock).toHaveBeenCalledWith( + '/api/copilot/confirm', + expect.objectContaining({ + body: expect.stringContaining('COPILOT_WORKFLOW_TOOL_BINDING_WORKFLOW_MISMATCH'), + }) + ) + }) + const confirmBody = JSON.parse( + fetchMock.mock.calls.find(([url]) => url === '/api/copilot/confirm')?.[1]?.body as string + ) + expect(confirmBody.message).toBe( + 'This Copilot workflow tool call is bound to a different workflow' + ) + expect(confirmBody.status).toBe('error') + }) + it('drops a duplicate async launch without confirming or surfacing an error', async () => { // The server fallback (or another tab) already claimed this tool call. // Reporting an error here would overwrite a run that is in flight. diff --git a/apps/sim/lib/copilot/tools/client/run-tool-execution.ts b/apps/sim/lib/copilot/tools/client/run-tool-execution.ts index 6ccb62f1f9..ac2ead7670 100644 --- a/apps/sim/lib/copilot/tools/client/run-tool-execution.ts +++ b/apps/sim/lib/copilot/tools/client/run-tool-execution.ts @@ -681,11 +681,20 @@ async function doExecuteRunTool( logger.error('[RunTool] Workflow execution threw', { toolCallId, toolName, error: msg }) const failedExecutionId = useExecutionStore.getState().getCurrentExecutionId(targetWorkflowId) ?? executionId + // Carry the real failure through instead of the generic "Workflow execution + // failed." — the agent can only correct a bad request (a rejected binding, + // an undeployed workflow) if it is told what was wrong. + const failureCode = isExecutionStreamHttpError(err) ? err.code : undefined await reportCompletion( toolCallId, MothershipStreamV1ToolOutcome.error, - getWorkflowToolCompletionMessage(MothershipStreamV1ToolOutcome.error), - undefined, + msg, + { + success: false, + workflowId: targetWorkflowId, + error: msg, + ...(failureCode ? { code: failureCode } : {}), + }, failedExecutionId ) } diff --git a/apps/sim/lib/copilot/tools/workflow-tools.test.ts b/apps/sim/lib/copilot/tools/workflow-tools.test.ts new file mode 100644 index 0000000000..b461634bad --- /dev/null +++ b/apps/sim/lib/copilot/tools/workflow-tools.test.ts @@ -0,0 +1,86 @@ +/** + * @vitest-environment node + */ + +import { describe, expect, it } from 'vitest' +import { + COPILOT_WORKFLOW_TOOL_BINDING_ERRORS, + classifyWorkflowToolBinding, + resolveWorkflowToolTargetId, +} from './workflow-tools' + +const runningToolCall = { + toolName: 'run_workflow', + status: 'running' as const, + permissionDecision: null, + args: { workflowId: 'workflow-1' }, +} + +const run = { userId: 'user-1', workflowId: 'workflow-1' } + +function classify(overrides: Partial[0]> = {}) { + return classifyWorkflowToolBinding({ + toolCall: runningToolCall, + run, + userId: 'user-1', + workflowId: 'workflow-1', + ...overrides, + }) +} + +describe('classifyWorkflowToolBinding', () => { + it('accepts a live call bound to the requested workflow', () => { + expect(classify()).toEqual({ ok: true }) + }) + + it.each([ + ['missing row', { toolCall: null }, COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.unknown], + [ + 'non-workflow tool', + { toolCall: { ...runningToolCall, toolName: 'read' } }, + COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.notWorkflowTool, + ], + [ + 'finished call', + { toolCall: { ...runningToolCall, status: 'completed' as const } }, + COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.alreadySettled, + ], + [ + 'unapproved call', + { toolCall: { ...runningToolCall, status: 'pending' as const } }, + COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.awaitingPermission, + ], + [ + 'another user', + { run: { ...run, userId: 'someone-else' } }, + COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.foreignOwner, + ], + [ + 'another workflow', + { workflowId: 'workflow-2' }, + COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.workflowMismatch, + ], + ])('rejects %s with its own reason', (_name, overrides, expected) => { + expect(classify(overrides)).toEqual({ ok: false, rejection: expected }) + }) + + it('reports a finished call as the same conflict the execution claim uses', () => { + // The client already treats this status/code pair as benign on both the sync + // and async paths, so a duplicate stops rendering as a workflow failure. + expect(COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.alreadySettled).toMatchObject({ + statusCode: 409, + code: 'COPILOT_WORKFLOW_EXECUTION_CONFLICT', + }) + }) + + it('falls back to the run workflow only for rows persisted without a stamped target', () => { + // Kept for the deploy window: in-flight tool calls created before the target + // was stamped into args still resolve through the run. + expect(resolveWorkflowToolTargetId({}, 'workflow-1')).toBe('workflow-1') + expect(classify({ toolCall: { ...runningToolCall, args: {} } })).toEqual({ ok: true }) + // A workspace chat has no run workflow, so nothing can rescue it. + expect( + classify({ toolCall: { ...runningToolCall, args: {} }, run: { ...run, workflowId: null } }) + ).toEqual({ ok: false, rejection: COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.workflowMismatch }) + }) +}) diff --git a/apps/sim/lib/copilot/tools/workflow-tools.ts b/apps/sim/lib/copilot/tools/workflow-tools.ts index b493f55c19..a922924b7e 100644 --- a/apps/sim/lib/copilot/tools/workflow-tools.ts +++ b/apps/sim/lib/copilot/tools/workflow-tools.ts @@ -1,8 +1,12 @@ +import type { CopilotAsyncToolStatus, CopilotToolPermissionDecision } from '@sim/db/schema' import { isPlainRecord } from '@sim/utils/object' import { ASYNC_TOOL_CONFIRMATION_STATUS, type AsyncConfirmationStatus, + isTerminalAsyncStatus, + isWorkflowToolExecutionClaimable, } from '@/lib/copilot/async-runs/lifecycle' +import { COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE } from '@/lib/copilot/constants' const WORKFLOW_TOOL_NAMES = [ 'run_workflow', @@ -31,6 +35,103 @@ const ASYNC_WORKFLOW_DEPLOYMENT_ERROR_BY_CODE = new Map [error.code, error]) ) +/** + * Why a workflow-tool execution request is not bound to the tool call it claims. + * + * These used to collapse into one opaque 403, which cost the caller any chance of + * telling "someone already ran this" (benign) from "this can never run" (a real + * defect), and told the model nothing it could act on. + * + * `alreadySettled` deliberately reuses `COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE`: + * it IS the same conflict as losing the execution claim, and both client paths + * already treat that status/code pair as benign and silent. + */ +export const COPILOT_WORKFLOW_TOOL_BINDING_ERRORS = { + unknown: { + code: 'COPILOT_WORKFLOW_TOOL_BINDING_UNKNOWN', + message: 'No Copilot workflow tool call matches this execution request', + statusCode: 404, + }, + notWorkflowTool: { + code: 'COPILOT_WORKFLOW_TOOL_BINDING_NOT_WORKFLOW_TOOL', + message: 'This Copilot tool call does not run a workflow', + statusCode: 403, + }, + alreadySettled: { + code: COPILOT_WORKFLOW_EXECUTION_CONFLICT_CODE, + message: 'This Copilot workflow tool call has already completed', + statusCode: 409, + }, + awaitingPermission: { + code: 'COPILOT_WORKFLOW_TOOL_BINDING_AWAITING_APPROVAL', + message: 'This Copilot workflow tool call has not been approved yet', + statusCode: 403, + }, + foreignOwner: { + code: 'COPILOT_WORKFLOW_TOOL_BINDING_FOREIGN_OWNER', + message: 'This Copilot workflow tool call belongs to a different user', + statusCode: 403, + }, + workflowMismatch: { + code: 'COPILOT_WORKFLOW_TOOL_BINDING_WORKFLOW_MISMATCH', + message: 'This Copilot workflow tool call is bound to a different workflow', + statusCode: 403, + }, +} as const + +export type CopilotWorkflowToolBindingError = + (typeof COPILOT_WORKFLOW_TOOL_BINDING_ERRORS)[keyof typeof COPILOT_WORKFLOW_TOOL_BINDING_ERRORS] + +export type CopilotWorkflowToolBindingResult = + | { ok: true } + | { ok: false; rejection: CopilotWorkflowToolBindingError } + +interface WorkflowToolBindingCandidate { + toolName: string + status: CopilotAsyncToolStatus + permissionDecision: CopilotToolPermissionDecision | null + args: unknown +} + +/** + * Decides whether an execution request may run under a Copilot workflow tool call. + * + * Non-authoritative on its own — the single-winner claim in + * `claimWorkflowToolExecution` is what actually prevents a double run. This exists + * so the request fails fast, and with a distinguishable reason, before spending + * admission and billing work on something that cannot legally run. + */ +export function classifyWorkflowToolBinding(params: { + toolCall: WorkflowToolBindingCandidate | null | undefined + run: { userId: string; workflowId: string | null } | null | undefined + userId: string + workflowId: string +}): CopilotWorkflowToolBindingResult { + const { toolCall, run, userId, workflowId } = params + const reject = (rejection: CopilotWorkflowToolBindingError) => ({ ok: false as const, rejection }) + + if (!toolCall) return reject(COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.unknown) + if (!isWorkflowToolName(toolCall.toolName)) { + return reject(COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.notWorkflowTool) + } + if (!isWorkflowToolExecutionClaimable(toolCall.status, toolCall.permissionDecision)) { + // Split the one unclaimable bucket: a finished call is a benign duplicate, + // an unapproved one is a real refusal. + return reject( + isTerminalAsyncStatus(toolCall.status) + ? COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.alreadySettled + : COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.awaitingPermission + ) + } + if (!run || run.userId !== userId) { + return reject(COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.foreignOwner) + } + if (resolveWorkflowToolTargetId(toolCall.args, run.workflowId) !== workflowId) { + return reject(COPILOT_WORKFLOW_TOOL_BINDING_ERRORS.workflowMismatch) + } + return { ok: true } +} + export function isWorkflowToolName(name: string): boolean { return WORKFLOW_TOOL_NAME_SET.has(name) }