Skip to content

Commit 83d4c9c

Browse files
committed
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.
1 parent 743daea commit 83d4c9c

7 files changed

Lines changed: 295 additions & 42 deletions

File tree

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

Lines changed: 31 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,13 +30,19 @@ import {
3030
presentWorkflowLogsForModel,
3131
} from '@/lib/mothership/tools/workflow-output'
3232
import {
33+
buildPausedWorkflowToolCompletion,
3334
createStructuralWorkflowToolCompletionData,
3435
getWorkflowToolCompletionExecutionId,
3536
getWorkflowToolCompletionMessage,
3637
getWorkflowToolLaunchError,
38+
WORKFLOW_TOOL_PAUSED_STATUS,
3739
type WorkflowToolLaunchError,
3840
} from '@/lib/mothership/tools/workflow-tools'
39-
import { getTrustedWorkflowToolExecution } from '@/lib/workflows/executor/execution-state'
41+
import {
42+
getHumanPauseResumeUrl,
43+
getTrustedWorkflowToolExecution,
44+
getWorkflowExecutionCopilotToolCallId,
45+
} from '@/lib/workflows/executor/execution-state'
4046
import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry'
4147

4248
const logger = createLogger('CopilotClientToolWaiter')
@@ -364,6 +370,30 @@ export async function waitForWorkflowToolCompletion({
364370
return structuralWorkflowCompletion(structuralStatus, workflowId, executionId, launchError)
365371
}
366372

373+
if (
374+
completion.status === MothershipStreamV1ToolOutcome.success &&
375+
isPlainRecord(completion.data) &&
376+
completion.data.status === WORKFLOW_TOOL_PAUSED_STATUS
377+
) {
378+
// Re-read rather than trust the confirmation; a run resumed since then reads as finished.
379+
const [resumeUrl, boundToolCallId] = await Promise.all([
380+
getHumanPauseResumeUrl(executionId, workflowId),
381+
getWorkflowExecutionCopilotToolCallId(executionId, workflowId),
382+
]).catch((error) => {
383+
logger.warn('Failed to read paused workflow tool execution', {
384+
toolCallId,
385+
workflowId,
386+
executionId,
387+
error: getErrorMessage(error),
388+
})
389+
return [null, undefined] as const
390+
})
391+
if (resumeUrl && boundToolCallId === toolCallId) {
392+
toolRegistry?.markIncomplete('client-tool-completion-deferred')
393+
return buildPausedWorkflowToolCompletion(workflowId, executionId, resumeUrl)
394+
}
395+
}
396+
367397
try {
368398
trustedExecution = await getTrustedWorkflowToolExecution(executionId, workflowId, toolCallId)
369399
} 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: 120 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,104 @@ 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("never hands a call the link to another call's paused run", async () => {
534+
const other = await claimedAndPausedUnreported('human')
535+
const { toolCallId } = await claimed()
536+
const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS })
537+
const claimedPause = { status: 'paused', workflowId, executionId: other.executionId }
538+
539+
await completeAsyncToolCall({ toolCallId, status: 'completed', result: claimedPause })
540+
publishToolConfirmation({
541+
toolCallId,
542+
status: 'success',
543+
timestamp: new Date().toISOString(),
544+
data: claimedPause,
545+
executionId: other.executionId,
546+
})
547+
548+
const completion = await waiting
549+
expect(completion?.data).not.toHaveProperty('resumeUrl')
550+
expect(JSON.stringify(completion)).not.toContain('/resume/')
551+
})
552+
553+
it('answers from the run log when a confirmation claims a pause the run is not in', async () => {
554+
const { toolCallId, executionId } = await claimedAndSettled()
555+
const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS })
556+
const claimedPause = {
557+
status: 'paused',
558+
workflowId,
559+
executionId,
560+
resumeUrl: 'https://elsewhere.test',
561+
}
562+
563+
await completeAsyncToolCall({ toolCallId, status: 'completed', result: claimedPause })
564+
publishToolConfirmation({
565+
toolCallId,
566+
status: 'success',
567+
timestamp: new Date().toISOString(),
568+
data: claimedPause,
569+
executionId,
570+
})
571+
572+
const completion = await waiting
573+
expect(completion?.status).toBe('success')
574+
expect(completion?.data).toEqual({ success: true, workflowId, executionId })
575+
})
576+
464577
it('moves a queued async run to the background when the browser never reports', async () => {
465578
const { toolCallId, executionId } = await claimed({ workflowId, async: true })
466579
const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS })

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

Lines changed: 66 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
import {
22
ASYNC_TOOL_CONFIRMATION_STATUS,
3+
type AsyncConfirmationStatus,
4+
type AsyncTerminalStatus,
35
getTerminalConfirmationStatus,
46
isTerminalAsyncStatus,
57
} from '@/lib/mothership/async-runs/lifecycle'
@@ -9,17 +11,50 @@ import {
911
} from '@/lib/mothership/async-runs/repository'
1012
import { publishToolConfirmation } from '@/lib/mothership/persistence/tool-confirm'
1113
import {
14+
buildPausedWorkflowToolCompletion,
1215
createStructuralWorkflowToolCompletionData,
1316
getWorkflowToolCompletionMessage,
1417
} from '@/lib/mothership/tools/workflow-tools'
15-
import { getWorkflowExecutionLogStatus } from '@/lib/workflows/executor/execution-state'
18+
import {
19+
getHumanPauseResumeUrl,
20+
getWorkflowExecutionLogStatus,
21+
} from '@/lib/workflows/executor/execution-state'
1622

1723
interface ReportClientWorkflowToolParams {
1824
toolCallId: string
1925
executionId: string
2026
workflowId: string
2127
}
2228

29+
/**
30+
* Completes the call the execution claimed and publishes its answer. Only the first transition
31+
* out of `running` lands; returns whether this one did.
32+
*/
33+
async function completeAndPublish(
34+
{ toolCallId, executionId }: ReportClientWorkflowToolParams,
35+
rowStatus: AsyncTerminalStatus,
36+
answer: { status: AsyncConfirmationStatus; message: string; data: Record<string, unknown> }
37+
): Promise<boolean> {
38+
const completed = await completeClientWorkflowToolCall(
39+
{
40+
toolCallId,
41+
status: rowStatus,
42+
result: answer.data,
43+
error: rowStatus === 'completed' ? null : answer.message,
44+
},
45+
executionId
46+
)
47+
if (!completed) return false
48+
49+
publishToolConfirmation({
50+
toolCallId,
51+
...answer,
52+
timestamp: new Date().toISOString(),
53+
executionId,
54+
})
55+
return true
56+
}
57+
2358
/**
2459
* Report a browser-claimed workflow tool's outcome from the execution it bound.
2560
*
@@ -32,39 +67,44 @@ interface ReportClientWorkflowToolParams {
3267
* started, which is what the browser reports from its stream error. Whichever of
3368
* this and the browser's report lands first is the one kept.
3469
*/
35-
export async function reportSettledClientWorkflowTool({
36-
toolCallId,
37-
executionId,
38-
workflowId,
39-
}: ReportClientWorkflowToolParams): Promise<void> {
70+
export async function reportSettledClientWorkflowTool(
71+
params: ReportClientWorkflowToolParams
72+
): Promise<void> {
73+
const { executionId, workflowId } = params
4074
const logStatus = await getWorkflowExecutionLogStatus(executionId, workflowId)
41-
if (logStatus !== undefined && !isTerminalAsyncStatus(logStatus)) return
75+
if (logStatus !== undefined && !isTerminalAsyncStatus(logStatus)) {
76+
await reportPausedClientWorkflowTool(params)
77+
return
78+
}
4279

4380
const executionStatus = logStatus ?? 'failed'
4481
const status = getTerminalConfirmationStatus(executionStatus)
45-
const message = getWorkflowToolCompletionMessage(status)
46-
const data = createStructuralWorkflowToolCompletionData(status, workflowId, executionId)
47-
const completed = await completeClientWorkflowToolCall(
48-
{
49-
toolCallId,
50-
status: executionStatus,
51-
result: data,
52-
error: executionStatus === 'completed' ? null : message,
53-
},
54-
executionId
55-
)
56-
if (!completed) return
57-
58-
publishToolConfirmation({
59-
toolCallId,
82+
await completeAndPublish(params, executionStatus, {
6083
status,
61-
message,
62-
timestamp: new Date().toISOString(),
63-
data,
64-
executionId,
84+
message: getWorkflowToolCompletionMessage(status),
85+
data: createStructuralWorkflowToolCompletionData(status, workflowId, executionId),
6586
})
6687
}
6788

89+
/**
90+
* Settle a browser-claimed workflow tool whose run is waiting on a person with the link to
91+
* answer, so Chat tells the user instead of waiting on an answer nobody knows is owed. A run
92+
* paused only on time resumes itself and keeps the call held. The call's claim on this
93+
* execution is what binds them, so no other call can be answered here. Returns whether it settled.
94+
*/
95+
export async function reportPausedClientWorkflowTool(
96+
params: ReportClientWorkflowToolParams
97+
): Promise<boolean> {
98+
const { executionId, workflowId } = params
99+
const resumeUrl = await getHumanPauseResumeUrl(executionId, workflowId)
100+
if (!resumeUrl) return false
101+
return completeAndPublish(
102+
params,
103+
'completed',
104+
buildPausedWorkflowToolCompletion(workflowId, executionId, resumeUrl)
105+
)
106+
}
107+
68108
/**
69109
* Move a browser-claimed async run to the background once the execute route has
70110
* queued it, the same transition the browser reports after the queue accepts

0 commit comments

Comments
 (0)