Make async tool resume delivery recoverable

This commit is contained in:
Siddharth Ganesan
2026-08-12 15:45:09 -07:00
parent a63c8fa35a
commit b5bae34edf
21 changed files with 910 additions and 126 deletions
@@ -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 }), {
@@ -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<boolean> {
}): Promise<CopilotWorkflowToolBindingResult> {
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) {
@@ -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) {
@@ -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',
@@ -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,
@@ -4145,14 +4145,14 @@ export const TOOL_RUNTIME_SCHEMAS: Record<string, ToolRuntimeSchemaEntry> = {
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<string, ToolRuntimeSchemaEntry> = {
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<string, ToolRuntimeSchemaEntry> = {
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<string, ToolRuntimeSchemaEntry> = {
"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<string, ToolRuntimeSchemaEntry> = {
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<string, ToolRuntimeSchemaEntry> = {
"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,
},
@@ -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',
@@ -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',
@@ -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<string, unknown> }).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 } })
@@ -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<ReturnType<typeof sealClientToolContext>> | 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<AsyncCompletionSignal> => {
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()
}
+27 -5
View File
@@ -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')
@@ -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()
@@ -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<Record<string, string>> = []
const executionContext: ExecutionContext = {
userId: 'user-1',
workflowId: '',
workspaceId: 'ws-1',
chatId: 'chat-1',
}
mockRunStreamLoop.mockImplementationOnce(
async (
_fetchUrl: string,
fetchOptions: RequestInit,
context: StreamingContext
): Promise<void> => {
headers.push(fetchOptions.headers as Record<string, string>)
context.errors.push(STREAM_ENDED_WITHOUT_TERMINAL_MESSAGE)
throw new StreamEndedWithoutTerminalError('/api/mothership')
}
)
mockRunStreamLoop.mockImplementationOnce(
async (
_fetchUrl: string,
fetchOptions: RequestInit,
context: StreamingContext
): Promise<void> => {
headers.push(fetchOptions.headers as Record<string, string>)
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<Record<string, unknown>> = []
mockRunStreamLoop.mockImplementation(
async (
fetchUrl: string,
fetchOptions: RequestInit,
context: StreamingContext
): Promise<void> => {
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<void> => {
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)
})
})
+152 -63
View File
@@ -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<string, string> {
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<AsyncContinuation | null> {
@@ -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<string, unknown> = 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<void> {
if (!abortSignal) {
return sleep(ms)
+15
View File
@@ -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.
@@ -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.
@@ -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,
@@ -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.
@@ -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
)
}
@@ -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<Parameters<typeof classifyWorkflowToolBinding>[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 })
})
})
@@ -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<string, AsyncWorkflowDep
Object.values(ASYNC_WORKFLOW_DEPLOYMENT_ERRORS).map((error) => [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)
}