Skip to content

Commit 89c1d63

Browse files
authored
fix(mothership): answer a Chat workflow tool paused for a person with its resume link (#9751)
* fix(mothership): answer a Chat workflow tool paused for a person with its resume link A browser-run workflow tool whose run paused at a Human in the Loop block was never answered: the browser reports nothing for a paused run, the execute route's report skipped a non-terminal log, and the lapsed-wait watcher only detaches a running log. The Chat turn parked on the call until it timed out, and the model never learned the run was waiting on the user or where to answer. - Settle the call when its run waits on a person, with the resume page link and an instruction to hand it to the user rather than resume it. A run paused only on time resumes itself and keeps the call held. - The Chat-side waiter re-reads the pause and the call's binding instead of trusting a confirmation's data. - The lapsed-wait watcher answers a human pause if the route's report failed. - Share the pause point parser between run status reads and this check. * fix(mothership): answer a paused Chat workflow tool still running when its form was submitted before Chat read the pause
1 parent 1bf1b85 commit 89c1d63

7 files changed

Lines changed: 347 additions & 42 deletions

File tree

‎apps/sim/lib/mothership/request/tools/client.ts‎

Lines changed: 45 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import {
66
type AsyncTerminalCompletionSnapshot,
77
getTerminalConfirmationStatus,
88
isAsyncTerminalConfirmationStatus,
9+
isTerminalAsyncStatus,
910
} from '@/lib/mothership/async-runs/lifecycle'
1011
import { replaceTerminalAsyncToolCallResult } from '@/lib/mothership/async-runs/repository'
1112
import { MothershipStreamV1ToolOutcome } from '@/lib/mothership/generated/mothership-stream-v1'
@@ -30,13 +31,20 @@ import {
3031
presentWorkflowLogsForModel,
3132
} from '@/lib/mothership/tools/workflow-output'
3233
import {
34+
buildPausedWorkflowToolCompletion,
3335
createStructuralWorkflowToolCompletionData,
3436
getWorkflowToolCompletionExecutionId,
3537
getWorkflowToolCompletionMessage,
3638
getWorkflowToolLaunchError,
39+
WORKFLOW_TOOL_PAUSED_STATUS,
3740
type WorkflowToolLaunchError,
3841
} from '@/lib/mothership/tools/workflow-tools'
39-
import { getTrustedWorkflowToolExecution } from '@/lib/workflows/executor/execution-state'
42+
import {
43+
getHumanPauseResumeUrl,
44+
getTrustedWorkflowToolExecution,
45+
getWorkflowExecutionCopilotToolCallId,
46+
getWorkflowExecutionLogStatus,
47+
} from '@/lib/workflows/executor/execution-state'
4048
import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
4149

4250
const logger = createLogger('CopilotClientToolWaiter')
@@ -364,6 +372,42 @@ export async function waitForWorkflowToolCompletion({
364372
return structuralWorkflowCompletion(structuralStatus, workflowId, executionId, launchError)
365373
}
366374

375+
if (
376+
completion.status === MothershipStreamV1ToolOutcome.success &&
377+
isPlainRecord(completion.data) &&
378+
completion.data.status === WORKFLOW_TOOL_PAUSED_STATUS
379+
) {
380+
// Re-read rather than trust the confirmation; a run resumed since then reads as finished.
381+
const [resumeUrl, boundToolCallId, logStatus] = await Promise.all([
382+
getHumanPauseResumeUrl(executionId, workflowId),
383+
getWorkflowExecutionCopilotToolCallId(executionId, workflowId),
384+
getWorkflowExecutionLogStatus(executionId, workflowId),
385+
]).catch((error) => {
386+
logger.warn('Failed to read paused workflow tool execution', {
387+
toolCallId,
388+
workflowId,
389+
executionId,
390+
error: getErrorMessage(error),
391+
})
392+
return [null, undefined, undefined] as const
393+
})
394+
if (boundToolCallId === toolCallId) {
395+
if (resumeUrl) {
396+
toolRegistry?.markIncomplete('client-tool-completion-deferred')
397+
return buildPausedWorkflowToolCompletion(workflowId, executionId, resumeUrl)
398+
}
399+
// The person answered before this read and the resumed run has no result yet.
400+
if (logStatus !== undefined && !isTerminalAsyncStatus(logStatus)) {
401+
toolRegistry?.markIncomplete('client-tool-completion-deferred')
402+
return structuralWorkflowCompletion(
403+
ASYNC_TOOL_CONFIRMATION_STATUS.background,
404+
workflowId,
405+
executionId
406+
)
407+
}
408+
}
409+
}
410+
367411
try {
368412
trustedExecution = await getTrustedWorkflowToolExecution(executionId, workflowId, toolCallId)
369413
} catch (error) {

‎apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import { recordDegraded } from '@/lib/mothership/request/metrics'
2020
import { waitForWorkflowToolCompletion } from '@/lib/mothership/request/tools/client'
2121
import { settleClientToolCall } from '@/lib/mothership/request/tools/client-settlement.server'
2222
import { toolStartWindowClosed } from '@/lib/mothership/request/tools/deadline'
23+
import { reportPausedClientWorkflowTool } from '@/lib/mothership/request/tools/workflow-client-settlement'
2324
import {
2425
createStructuralWorkflowToolCompletionData,
2526
getWorkflowToolCompletionMessage,
@@ -98,6 +99,16 @@ async function detachClientRunAfterWait(params: {
9899
const logged = await getWorkflowExecutionLogStatus(executionId, workflowId).catch(
99100
() => undefined
100101
)
102+
// A paused run's log stays `pending`. The execute route answers a run waiting on a person;
103+
// this covers that best-effort report failing.
104+
if (logged === 'pending') {
105+
const answered = await reportPausedClientWorkflowTool({
106+
toolCallId,
107+
executionId,
108+
workflowId,
109+
}).catch(() => false)
110+
if (answered) return
111+
}
101112
if (logged !== 'running') continue
102113
const status = ASYNC_TOOL_CONFIRMATION_STATUS.background
103114
const settled = await settleClientToolCall({

‎apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts‎

Lines changed: 158 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,10 @@ import {
4242
settleClientWorkflowToolExecution,
4343
upsertAsyncToolCall,
4444
} from '@/lib/mothership/async-runs/repository'
45-
import { getToolConfirmation } from '@/lib/mothership/persistence/tool-confirm'
45+
import {
46+
getToolConfirmation,
47+
publishToolConfirmation,
48+
} from '@/lib/mothership/persistence/tool-confirm'
4649
import { waitForWorkflowToolCompletion } from '@/lib/mothership/request/tools/client'
4750
import { raceWorkflowToolClientPickup } from '@/lib/mothership/request/tools/workflow-client-fallback'
4851
import {
@@ -207,10 +210,13 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () =>
207210
}
208211

209212
/**
210-
* A claimed call whose bound execution paused at a human approval: the execute route settled the
211-
* claim and its own report left the call running, as it does for every paused execution.
213+
* A claimed call whose bound execution paused (a paused run's log stays `pending`) and whose
214+
* execute route settled the claim, before the route reports it.
212215
*/
213-
async function claimedAndPaused(snapshot: string = UNRESUMABLE_SNAPSHOT) {
216+
async function claimedAndPausedUnreported(
217+
pauseKind: 'human' | 'time',
218+
snapshot: string = UNRESUMABLE_SNAPSHOT
219+
) {
214220
const { toolCallId, executionId } = await claimed()
215221
await executionLog(toolCallId, executionId, 'pending')
216222
await db.insert(pausedExecutions).values({
@@ -226,17 +232,26 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () =>
226232
registeredAt: new Date().toISOString(),
227233
resumeStatus: 'paused',
228234
snapshotReady: true,
229-
pauseKind: 'human',
235+
pauseKind,
230236
},
231237
},
232238
totalPauseCount: 1,
233239
})
234240
await settleClientWorkflowToolExecution(toolCallId, executionId)
235-
await reportSettledClientWorkflowTool({ toolCallId, executionId, workflowId })
236-
expect(await toolRow(toolCallId)).toMatchObject({ status: 'running', result: null })
237241
return { toolCallId, executionId }
238242
}
239243

244+
/**
245+
* A claimed call whose bound execution paused until a scheduled time: the run resumes by
246+
* itself, so the execute route's report leaves the call held for the resume to settle.
247+
*/
248+
async function claimedAndPaused(snapshot: string = UNRESUMABLE_SNAPSHOT) {
249+
const ids = await claimedAndPausedUnreported('time', snapshot)
250+
await reportSettledClientWorkflowTool({ ...ids, workflowId })
251+
expect(await toolRow(ids.toolCallId)).toMatchObject({ status: 'running', result: null })
252+
return ids
253+
}
254+
240255
async function setLogStatus(executionId: string, status: 'completed' | 'failed' | 'cancelled') {
241256
await db
242257
.update(workflowExecutionLogs)
@@ -461,6 +476,142 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () =>
461476
expect(await toolRow(toolCallId)).toMatchObject({ status: 'failed', claimedBy: null })
462477
})
463478

479+
it('answers a run paused for a person with the link to its form, and keeps that answer when the run is later cancelled', async () => {
480+
const { toolCallId, executionId } = await claimedAndPausedUnreported('human')
481+
const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS })
482+
483+
await reportSettledClientWorkflowTool({ toolCallId, executionId, workflowId })
484+
485+
const paused = {
486+
success: true,
487+
status: 'paused',
488+
workflowId,
489+
executionId,
490+
resumeUrl: `http://localhost:3000/resume/${workflowId}/${executionId}`,
491+
}
492+
const completion = await waiting
493+
expect(completion).toMatchObject({ status: 'success', data: paused })
494+
expect(completion?.message).toContain(paused.resumeUrl)
495+
const settled = await toolRow(toolCallId)
496+
expect(settled).toMatchObject({ status: 'completed', claimedBy: null, result: paused })
497+
expect(await publishedConfirmation(toolCallId)).toMatchObject({
498+
status: 'success',
499+
executionId,
500+
})
501+
502+
// The run is cancelled before anyone submits: its own report finds the call already answered.
503+
expect(await PauseResumeManager.beginPausedCancellation(executionId, workflowId)).toBe(true)
504+
expect(await PauseResumeManager.completePausedCancellation(executionId, workflowId)).toBe(true)
505+
expect(await toolRow(toolCallId)).toMatchObject({
506+
status: 'completed',
507+
completedAt: settled.completedAt,
508+
result: paused,
509+
})
510+
})
511+
512+
it("answers a paused run the worker's lapsed wait finds when the route's report never landed", async () => {
513+
const { toolCallId, executionId } = await claimedAndPausedUnreported('human')
514+
const runOnServer = vi.fn()
515+
516+
const outcome = await raceWorkflowToolClientPickup({
517+
toolCallId,
518+
workflowId,
519+
timeoutMs: WAIT_MS,
520+
graceMs: WAIT_MS,
521+
waitTimeoutMs: 500,
522+
runOnServer,
523+
})
524+
525+
expect(outcome).toMatchObject({
526+
winner: 'client',
527+
completion: { status: 'success', data: { status: 'paused', workflowId, executionId } },
528+
})
529+
expect(runOnServer).not.toHaveBeenCalled()
530+
expect(await toolRow(toolCallId)).toMatchObject({ status: 'completed', claimedBy: null })
531+
})
532+
533+
it('answers a paused call still running when the person submitted before Chat read the pause', async () => {
534+
const { toolCallId, executionId } = await claimedAndPausedUnreported('human')
535+
await reportSettledClientWorkflowTool({ toolCallId, executionId, workflowId })
536+
// The resume took the pause point and the resumed run is going.
537+
const [pausedRow] = await db
538+
.select({ id: pausedExecutions.id, pausePoints: pausedExecutions.pausePoints })
539+
.from(pausedExecutions)
540+
.where(eq(pausedExecutions.executionId, executionId))
541+
await db
542+
.update(pausedExecutions)
543+
.set({
544+
pausePoints: {
545+
[APPROVAL_CONTEXT_ID]: {
546+
...(pausedRow.pausePoints as Record<string, Record<string, unknown>>)[
547+
APPROVAL_CONTEXT_ID
548+
],
549+
resumeStatus: 'resuming',
550+
},
551+
},
552+
})
553+
.where(eq(pausedExecutions.id, pausedRow.id))
554+
await db
555+
.update(workflowExecutionLogs)
556+
.set({ status: 'running' })
557+
.where(eq(workflowExecutionLogs.executionId, executionId))
558+
559+
const completion = await waitForWorkflowToolCompletion({
560+
toolCallId,
561+
workflowId,
562+
timeoutMs: WAIT_MS,
563+
})
564+
565+
expect(completion).toMatchObject({
566+
status: 'background',
567+
data: { status: 'running', workflowId, executionId },
568+
})
569+
})
570+
571+
it("never hands a call the link to another call's paused run", async () => {
572+
const other = await claimedAndPausedUnreported('human')
573+
const { toolCallId } = await claimed()
574+
const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS })
575+
const claimedPause = { status: 'paused', workflowId, executionId: other.executionId }
576+
577+
await completeAsyncToolCall({ toolCallId, status: 'completed', result: claimedPause })
578+
publishToolConfirmation({
579+
toolCallId,
580+
status: 'success',
581+
timestamp: new Date().toISOString(),
582+
data: claimedPause,
583+
executionId: other.executionId,
584+
})
585+
586+
const completion = await waiting
587+
expect(completion?.data).not.toHaveProperty('resumeUrl')
588+
expect(JSON.stringify(completion)).not.toContain('/resume/')
589+
})
590+
591+
it('answers from the run log when a confirmation claims a pause the run is not in', async () => {
592+
const { toolCallId, executionId } = await claimedAndSettled()
593+
const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS })
594+
const claimedPause = {
595+
status: 'paused',
596+
workflowId,
597+
executionId,
598+
resumeUrl: 'https://elsewhere.test',
599+
}
600+
601+
await completeAsyncToolCall({ toolCallId, status: 'completed', result: claimedPause })
602+
publishToolConfirmation({
603+
toolCallId,
604+
status: 'success',
605+
timestamp: new Date().toISOString(),
606+
data: claimedPause,
607+
executionId,
608+
})
609+
610+
const completion = await waiting
611+
expect(completion?.status).toBe('success')
612+
expect(completion?.data).toEqual({ success: true, workflowId, executionId })
613+
})
614+
464615
it('moves a queued async run to the background when the browser never reports', async () => {
465616
const { toolCallId, executionId } = await claimed({ workflowId, async: true })
466617
const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS })

0 commit comments

Comments
 (0)