Skip to content

Commit b1aabef

Browse files
committed
fix(copilot): interrupt pending tool waits on stop
1 parent 5db71b8 commit b1aabef

2 files changed

Lines changed: 171 additions & 11 deletions

File tree

apps/sim/lib/copilot/request/lifecycle/run.test.ts

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2063,6 +2063,73 @@ describe('runCopilotLifecycle', () => {
20632063
}
20642064
})
20652065

2066+
it('cancels promptly while a sequential tool promise remains unsettled', async () => {
2067+
vi.useFakeTimers()
2068+
try {
2069+
const controller = new AbortController()
2070+
const fetchUrls: string[] = []
2071+
let capturedContext: StreamingContext | null = null
2072+
const executionContext: ExecutionContext = {
2073+
userId: 'user-1',
2074+
workflowId: '',
2075+
workspaceId: 'ws-1',
2076+
chatId: 'chat-1',
2077+
}
2078+
2079+
mockPendingToolWaitBudgetMs.mockReturnValue(3_600_000)
2080+
mockRunStreamLoop.mockImplementationOnce(
2081+
async (fetchUrl: string, _fetchOptions: RequestInit, context: StreamingContext) => {
2082+
fetchUrls.push(fetchUrl)
2083+
capturedContext = context
2084+
context.toolCalls.set('tool-hung', {
2085+
id: 'tool-hung',
2086+
name: 'terminal',
2087+
status: 'awaiting_approval',
2088+
})
2089+
context.pendingToolPromises.set('tool-hung', new Promise(() => {}))
2090+
context.awaitingAsyncContinuation = {
2091+
checkpointId: 'ckpt-1',
2092+
pendingToolCallIds: ['tool-hung'],
2093+
}
2094+
}
2095+
)
2096+
2097+
const lifecycle = runCopilotLifecycle(
2098+
{ message: 'hello', messageId: 'stream-aborted-tool-wait' },
2099+
{
2100+
userId: 'user-1',
2101+
workspaceId: 'ws-1',
2102+
chatId: 'chat-1',
2103+
executionId: 'exec-1',
2104+
runId: 'run-1',
2105+
executionContext,
2106+
abortSignal: controller.signal,
2107+
}
2108+
)
2109+
2110+
await vi.advanceTimersByTimeAsync(0)
2111+
expect(mockPendingToolWaitBudgetMs).toHaveBeenCalled()
2112+
controller.abort('user_stop')
2113+
await vi.advanceTimersByTimeAsync(0)
2114+
const result = await lifecycle
2115+
2116+
expect(result.success).toBe(false)
2117+
expect(result.cancelled).toBe(true)
2118+
expect(fetchUrls).toEqual(['http://mothership.test/api/copilot'])
2119+
expect(mockForceFailHungToolCall).not.toHaveBeenCalled()
2120+
expect(capturedContext?.toolCalls.get('tool-hung')).toMatchObject({
2121+
status: MothershipStreamV1ToolOutcome.cancelled,
2122+
error: 'Stopped by user',
2123+
})
2124+
2125+
await vi.advanceTimersByTimeAsync(3_700_000)
2126+
expect(mockForceFailHungToolCall).not.toHaveBeenCalled()
2127+
expect(fetchUrls).toEqual(['http://mothership.test/api/copilot'])
2128+
} finally {
2129+
vi.useRealTimers()
2130+
}
2131+
})
2132+
20662133
it('force-fails each hung tool on its own budget while awaiting a long approval', async () => {
20672134
vi.useFakeTimers()
20682135
try {
@@ -2420,6 +2487,57 @@ describe('runCopilotLifecycle', () => {
24202487
expect(result.success).toBe(true)
24212488
})
24222489

2490+
it('cancels promptly while a per-subagent tool promise remains unsettled', async () => {
2491+
const controller = new AbortController()
2492+
const addAbortListener = vi.spyOn(controller.signal, 'addEventListener')
2493+
const fetchUrls: string[] = []
2494+
let capturedContext: StreamingContext | null = null
2495+
mockRunStreamLoop.mockImplementationOnce(
2496+
async (fetchUrl: string, _fetchOptions: RequestInit, context: StreamingContext) => {
2497+
fetchUrls.push(fetchUrl)
2498+
capturedContext = context
2499+
context.toolCalls.set('tool-hung', {
2500+
id: 'tool-hung',
2501+
name: 'read',
2502+
status: 'executing',
2503+
})
2504+
context.pendingToolPromises.set('tool-hung', new Promise(() => {}))
2505+
context.awaitingAsyncContinuation = {
2506+
checkpointId: 'cp-root',
2507+
pendingToolCallIds: ['tool-hung'],
2508+
frames: [
2509+
{
2510+
parentToolCallId: 'subagent-file',
2511+
parentToolName: 'file',
2512+
pendingToolIds: ['tool-hung'],
2513+
checkpointId: 'cp-file',
2514+
},
2515+
],
2516+
}
2517+
}
2518+
)
2519+
2520+
const lifecycle = runCopilotLifecycle(
2521+
{ message: 'hello', messageId: 'stream-aborted-subagent-wait' },
2522+
{ userId: 'user-1', workspaceId: 'ws-1', abortSignal: controller.signal }
2523+
)
2524+
2525+
await vi.waitFor(() => {
2526+
expect(addAbortListener).toHaveBeenCalledWith('abort', expect.any(Function), { once: true })
2527+
})
2528+
controller.abort('user_stop')
2529+
const result = await lifecycle
2530+
2531+
expect(result.success).toBe(false)
2532+
expect(result.cancelled).toBe(true)
2533+
expect(fetchUrls).toEqual(['http://mothership.test/api/copilot'])
2534+
expect(mockForceFailHungToolCall).not.toHaveBeenCalled()
2535+
expect(capturedContext?.toolCalls.get('tool-hung')).toMatchObject({
2536+
status: MothershipStreamV1ToolOutcome.cancelled,
2537+
error: 'Stopped by user',
2538+
})
2539+
})
2540+
24232541
it('classifies a Stop landing during a subagent fanout as cancelled', async () => {
24242542
// Guards the trap in the fanout fix: `wasAborted` is now isolated per leg, so
24252543
// a user Stop must still reach the turn — via the abort signal or the folded

apps/sim/lib/copilot/request/lifecycle/run.ts

Lines changed: 53 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -538,13 +538,34 @@ export function mergeResumeLegOutputs(
538538
if (leg.completionStatus) context.completionStatus = leg.completionStatus
539539
}
540540

541-
async function waitForToolIds(context: StreamingContext, toolIds: string[]): Promise<void> {
541+
async function waitForToolIds(
542+
context: StreamingContext,
543+
toolIds: string[],
544+
abortSignal?: AbortSignal
545+
): Promise<boolean> {
542546
const promises: Promise<unknown>[] = []
543547
for (const id of toolIds) {
544548
const p = context.pendingToolPromises.get(id)
545549
if (p) promises.push(p)
546550
}
547-
if (promises.length > 0) await Promise.allSettled(promises)
551+
if (promises.length === 0) return true
552+
if (!abortSignal) {
553+
await Promise.allSettled(promises)
554+
return true
555+
}
556+
if (abortSignal.aborted) return false
557+
558+
let onAbort = () => {}
559+
const aborted = new Promise<false>((resolve) => {
560+
onAbort = () => resolve(false)
561+
abortSignal.addEventListener('abort', onAbort, { once: true })
562+
if (abortSignal.aborted) onAbort()
563+
})
564+
try {
565+
return await Promise.race([Promise.allSettled(promises).then(() => true as const), aborted])
566+
} finally {
567+
abortSignal.removeEventListener('abort', onAbort)
568+
}
548569
}
549570

550571
interface ResumeToolResult {
@@ -689,7 +710,8 @@ async function driveOneChildChain(
689710
for (;;) {
690711
if (isAborted(options, context)) return null
691712

692-
await waitForToolIds(context, toolIds)
713+
const toolsSettled = await waitForToolIds(context, toolIds, options.abortSignal)
714+
if (!toolsSettled || isAborted(options, context)) return null
693715
const results = collectResultsForToolIds(context, toolIds, checkpointId)
694716

695717
const leg = makeResumeLegContext(context)
@@ -1007,7 +1029,15 @@ async function runCheckpointLoop(
10071029
next = null
10081030
break
10091031
}
1010-
await waitForToolIds(context, next.pendingToolCallIds)
1032+
const toolsSettled = await waitForToolIds(
1033+
context,
1034+
next.pendingToolCallIds,
1035+
options.abortSignal
1036+
)
1037+
if (!toolsSettled || isAborted(options, context)) {
1038+
next = null
1039+
break
1040+
}
10111041
next = await driveSubagentChains(
10121042
next,
10131043
context,
@@ -1018,7 +1048,10 @@ async function runCheckpointLoop(
10181048
hostedBillingRequest
10191049
)
10201050
}
1021-
if (!next) break
1051+
if (!next) {
1052+
if (isAborted(options, context)) cancelPendingTools(context)
1053+
break
1054+
}
10221055
continuation = next
10231056
}
10241057

@@ -1049,6 +1082,7 @@ async function runCheckpointLoop(
10491082
* sibling. Unchanged promises retain their absolute deadlines.
10501083
*/
10511084
while (context.pendingToolPromises.size > 0) {
1085+
if (isAborted(options, context)) break
10521086
const now = Date.now()
10531087
for (const [toolCallId, watchdog] of pendingWatchdogs) {
10541088
if (context.pendingToolPromises.get(toolCallId) !== watchdog.promise) {
@@ -1116,28 +1150,36 @@ async function runCheckpointLoop(
11161150
})
11171151

11181152
const watchdogController = new AbortController()
1153+
const waitSignal = options.abortSignal
1154+
? AbortSignal.any([watchdogController.signal, options.abortSignal])
1155+
: watchdogController.signal
11191156
try {
11201157
const wake = await Promise.race([
11211158
...activeWatchdogs.map(([, watchdog]) => watchdog.settlement),
1122-
interruptibleSleep(
1123-
Math.max(0, nextDeadlineAt - Date.now()),
1124-
watchdogController.signal
1125-
).then(() => null),
1159+
interruptibleSleep(Math.max(0, nextDeadlineAt - Date.now()), waitSignal).then(
1160+
() => null
1161+
),
11261162
])
1163+
if (isAborted(options, context)) break
11271164
if (wake && context.pendingToolPromises.get(wake.toolCallId) === wake.promise) {
11281165
context.pendingToolPromises.delete(wake.toolCallId)
11291166
}
11301167
} finally {
11311168
watchdogController.abort()
11321169
}
11331170
}
1171+
const waitWasAborted = isAborted(options, context)
11341172
waitSpan.attributes = {
11351173
...waitSpan.attributes,
11361174
waitBudgetMs: maximumWaitBudgetMs,
11371175
timedOutCount,
1138-
settledInTime: timedOutCount === 0,
1176+
aborted: waitWasAborted,
1177+
settledInTime: timedOutCount === 0 && !waitWasAborted,
11391178
}
1140-
context.trace.endSpan(waitSpan)
1179+
context.trace.endSpan(
1180+
waitSpan,
1181+
waitWasAborted ? RequestTraceV1SpanStatus.cancelled : RequestTraceV1SpanStatus.ok
1182+
)
11411183
}
11421184

11431185
if (isAborted(options, context)) {

0 commit comments

Comments
 (0)