From 9b9dd8c395b3e0f216c93f37d791468c3417efe0 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 11:09:22 -0700 Subject: [PATCH 1/3] fix(mothership): Stop only cancels detached runs still running, and a held workflow run never blocks the next turn's workbench --- .../lib/mothership/async-runs/repository.ts | 63 +++++-- .../workflow-client-settlement.integration.ts | 163 ++++++++++++++++-- .../mothership/tools/client-executed-tools.ts | 7 +- .../tools/handlers/workflow/mutations.ts | 38 +++- 4 files changed, 234 insertions(+), 37 deletions(-) diff --git a/apps/sim/lib/mothership/async-runs/repository.ts b/apps/sim/lib/mothership/async-runs/repository.ts index bc441b9165f..f2b236f4ad9 100644 --- a/apps/sim/lib/mothership/async-runs/repository.ts +++ b/apps/sim/lib/mothership/async-runs/repository.ts @@ -11,6 +11,7 @@ import { copilotRequestStops, copilotRuns, user, + workflowExecutionLogs, } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { filterUndefined, toRecordOrNull } from '@sim/utils/object' @@ -79,6 +80,10 @@ import { NO_SECRET_AUTHORITY, type TurnSecretAuthority, } from '@/lib/mothership/secret-mount-policy' +import { + isWorkflowToolName, + WORKFLOW_TOOL_NAMES, +} from '@/lib/mothership/tools/client-executed-tools' import { type DesktopToolClaimOwner, isDesktopToolCall, @@ -1753,6 +1758,7 @@ export async function prepareWorkbenchAccess( settledAt: copilotAsyncToolCalls.executionSettledAt, revokedAt: copilotAsyncToolCalls.executionRevokedAt, toolCallId: copilotAsyncToolCalls.toolCallId, + executableName: copilotAsyncToolCalls.executableName, processes: copilotAsyncToolCalls.sandboxProcesses, }) .from(copilotAsyncToolCalls) @@ -1820,7 +1826,16 @@ export async function prepareWorkbenchAccess( if (processes.some((process) => process.sessionKey !== input.sessionKey)) throw new Error('Recorded workbench ownership does not match this chat') return { - handlersPending: previous.some((row) => !row.settledAt && !row.revokedAt), + /** + * A workflow tool's handler runs a workflow, which never touches the workbench. One answered + * still running holds its run past its turn, so it must not hold the workbench too. + */ + handlersPending: previous.some( + (row) => + !row.settledAt && + !row.revokedAt && + !(row.executableName && isWorkflowToolName(row.executableName)) + ), processes, } }) @@ -1925,13 +1940,6 @@ export async function getUnsettledClientWorkflowExecutions( ) } -const WORKFLOW_RUN_TOOL_NAMES = [ - 'run_workflow', - 'run_workflow_until_block', - 'run_from_block', - 'run_block', -] as const - /** * The workflow execution a settled call started and left running: a `workflows run` the CLI * answered still running (exit 4 and its run record), or a workflow tool answered still running @@ -1969,8 +1977,11 @@ export function detachedWorkflowExecutionId(call: { } /** - * Workflow executions this run's calls started and left running past the call: what a Stop - * cancels beyond the calls still open, since a detached run takes no call's abort signal. + * Workflow executions this run's calls started and left running past the call, and that are still + * running: what a Stop cancels beyond the calls still open, since a detached run takes no call's + * abort signal. A run whose log has settled or paused is left alone: a paused run waits on a + * person, not on this turn, and the durable cancel flag would fail its resume under the same + * execution id. A run with no log yet may still be starting, so it is cancelled. */ export async function getDetachedWorkflowExecutions( streamId: string, @@ -1983,11 +1994,13 @@ export async function getDetachedWorkflowExecutions( 'copilot_async_tool_calls', { [TraceAttr.StreamId]: streamId, [TraceAttr.UserId]: userId }, async () => { + const { args, result } = copilotAsyncToolCalls + /** Only the fields `detachedWorkflowExecutionId` reads, so no call's payload is loaded. */ const rows = await executor .select({ executableName: copilotAsyncToolCalls.executableName, - args: copilotAsyncToolCalls.args, - result: copilotAsyncToolCalls.result, + args: sql`jsonb_build_object('request', jsonb_build_object('invocation', jsonb_build_object('argv', ${args} #> '{request,invocation,argv}')))`, + result: sql`jsonb_build_object('exitCode', ${result} -> 'exitCode', 'stdout', ${result} -> 'stdout', 'status', ${result} -> 'status', 'executionId', ${result} -> 'executionId')`, }) .from(copilotAsyncToolCalls) .innerJoin(copilotRuns, eq(copilotRuns.id, copilotAsyncToolCalls.runId)) @@ -1995,11 +2008,31 @@ export async function getDetachedWorkflowExecutions( and( eq(copilotRuns.streamId, streamId), eq(copilotRuns.userId, userId), - inArray(copilotAsyncToolCalls.executableName, ['sim_cli', ...WORKFLOW_RUN_TOOL_NAMES]), - isNotNull(copilotAsyncToolCalls.result) + or( + and( + eq(copilotAsyncToolCalls.executableName, 'sim_cli'), + sql`${result} ->> 'exitCode' = '4'` + ), + and( + inArray(copilotAsyncToolCalls.executableName, [...WORKFLOW_TOOL_NAMES]), + sql`${result} ->> 'status' = 'running'` + ) + ) + ) + ) + const detached = [...new Set(rows.flatMap((row) => detachedWorkflowExecutionId(row) ?? []))] + if (detached.length === 0) return [] + const settled = await executor + .select({ executionId: workflowExecutionLogs.executionId }) + .from(workflowExecutionLogs) + .where( + and( + inArray(workflowExecutionLogs.executionId, detached), + ne(workflowExecutionLogs.status, 'running') ) ) - return [...new Set(rows.flatMap((row) => detachedWorkflowExecutionId(row) ?? []))] + const notRunning = new Set(settled.map((row) => row.executionId)) + return detached.filter((executionId) => !notRunning.has(executionId)) } ) } diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts b/apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts index 20d7806c949..a332e2fd007 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts @@ -33,11 +33,14 @@ import { eq, inArray } from 'drizzle-orm' import { closeRedisConnection, getRedisClient } from '@/lib/core/config/redis' import { SIM_TOOL_EXECUTION_VERSION } from '@/lib/mothership/async-runs/lifecycle' import { + claimToolExecution, claimWorkflowToolExecution, completeAsyncToolCall, detachAsyncToolCall, getDetachedWorkflowExecutions, + prepareWorkbenchAccess, settleClientWorkflowToolExecution, + upsertAsyncToolCall, } from '@/lib/mothership/async-runs/repository' import { getToolConfirmation } from '@/lib/mothership/persistence/tool-confirm' import { waitForWorkflowToolCompletion } from '@/lib/mothership/request/tools/client' @@ -46,6 +49,7 @@ import { reportQueuedClientWorkflowTool, reportSettledClientWorkflowTool, } from '@/lib/mothership/request/tools/workflow-client-settlement' +import { chatSandboxSessionKey } from '@/lib/mothership/tools/sandbox-session-key' import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager' /** Longer than the waiter's durable poll, far shorter than the hour it used to park for. */ @@ -532,24 +536,30 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => }) it("names the runs this run's calls started and left running, for a Stop to cancel", async () => { - const insertCall = async (executableName: string, args: unknown, result: unknown) => - db.insert(copilotAsyncToolCalls).values({ + const insertCall = async (executableName: string, args: unknown, result: unknown) => { + const toolCallId = generateId() + await db.insert(copilotAsyncToolCalls).values({ runId, - toolCallId: generateId(), + toolCallId, toolName: executableName, executableName, args: args as Record, result: result as Record, status: 'completed', }) - const detachedCli = generateId() - const detachedTool = generateId() + return toolCallId + } const record = (runIdValue: string, status: string) => JSON.stringify({ runId: runIdValue, workflowId, status }) - await insertCall( - 'sim_cli', - { request: { invocation: { kind: 'cli', argv: ['workflows', 'run', workflowId] } } }, - { exitCode: 4, stdout: record(detachedCli, 'running'), stderr: '' } + const detachedCli = generateId() + await executionLog( + await insertCall( + 'sim_cli', + { request: { invocation: { kind: 'cli', argv: ['workflows', 'run', workflowId] } } }, + { exitCode: 4, stdout: record(detachedCli, 'running'), stderr: '' } + ), + detachedCli, + 'running' ) // A wait on a run proves nothing about who started it. await insertCall( @@ -557,15 +567,144 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => { request: { invocation: { kind: 'cli', argv: ['workflows', 'runs', 'wait', 'other'] } } }, { exitCode: 4, stdout: record(generateId(), 'running'), stderr: '' } ) + const detachedTool = generateId() + await executionLog( + await insertCall( + 'run_workflow', + { workflowId }, + { status: 'running', executionId: detachedTool, workflowId } + ), + detachedTool, + 'running' + ) + // Answered running but not yet logged: it may still be starting, and a Stop must reach it. + const startingTool = generateId() await insertCall( - 'run_workflow', + 'run_from_block', { workflowId }, - { status: 'running', executionId: detachedTool, workflowId } + { status: 'running', executionId: startingTool, workflowId } ) await insertCall('run_block', { workflowId }, { success: true, executionId: generateId() }) expect((await getDetachedWorkflowExecutions(streamId, userId)).sort()).toEqual( - [detachedCli, detachedTool].sort() + [detachedCli, detachedTool, startingTool].sort() ) }) + + it.each(['pending', 'completed', 'failed', 'cancelled'] as const)( + 'leaves a detached run whose log is %s for a Stop to skip', + async (logStatus) => { + const detachedCli = generateId() + const detachedTool = generateId() + const cliCall = generateId() + const toolCall = generateId() + await db.insert(copilotAsyncToolCalls).values([ + { + runId, + toolCallId: cliCall, + toolName: 'sim_cli', + executableName: 'sim_cli', + args: { + request: { invocation: { kind: 'cli', argv: ['workflows', 'run', workflowId] } }, + }, + result: { + exitCode: 4, + stdout: JSON.stringify({ runId: detachedCli, workflowId, status: 'running' }), + stderr: '', + }, + status: 'completed', + }, + { + runId, + toolCallId: toolCall, + toolName: 'run_workflow', + executableName: 'run_workflow', + args: { workflowId }, + result: { status: 'running', executionId: detachedTool, workflowId }, + status: 'completed', + }, + ]) + // The run went on past its call and then settled, or paused at a human approval: a paused + // run's resume subscribes under the same id, so a cancel flag would fail it. + await executionLog(cliCall, detachedCli, logStatus) + await executionLog(toolCall, detachedTool, logStatus) + + const cancellable = await getDetachedWorkflowExecutions(streamId, userId) + expect(cancellable).not.toContain(detachedCli) + expect(cancellable).not.toContain(detachedTool) + } + ) + + /** + * An earlier turn of a new chat whose `earlierTool` call is still executing under a live lease, + * as a workflow tool answered still running is while its run goes on, and a later turn whose + * own Sim call asks for the chat's workbench. + */ + async function laterTurnWorkbenchAccess(earlierTool: string) { + const turnChatId = generateId() + await db.insert(copilotChats).values({ + id: turnChatId, + userId, + workspaceId, + type: 'mothership', + conversationId: generateId(), + }) + try { + const turn = async (startedAt: Date, toolName: string) => { + const turnRunId = generateId() + await db.insert(copilotRuns).values({ + id: turnRunId, + executionId: generateId(), + chatId: turnChatId, + userId, + workspaceId, + streamId: generateId(), + toolExecutionVersion: SIM_TOOL_EXECUTION_VERSION, + status: 'active', + startedAt, + requestContext: { source: 'headless_lifecycle' }, + }) + const owner = { + toolCallId: generateId(), + runId: turnRunId, + userId, + ownerToken: generateId(), + } + await upsertAsyncToolCall({ runId: turnRunId, toolCallId: owner.toolCallId, toolName }) + expect(await claimToolExecution(owner)).toMatchObject({ outcome: 'claimed' }) + return owner + } + const earlier = await turn(new Date(Date.now() - 60_000), earlierTool) + await completeAsyncToolCall({ + toolCallId: earlier.toolCallId, + status: 'completed', + result: { status: 'running', executionId: generateId(), workflowId }, + }) + // The earlier turn answered and ended; its call's execution still holds the run. + await db + .update(copilotRuns) + .set({ status: 'complete', toolAdmissionClosedAt: new Date() }) + .where(eq(copilotRuns.id, earlier.runId)) + const later = await turn(new Date(), 'function_execute') + return await prepareWorkbenchAccess({ + ...later, + sessionKey: chatSandboxSessionKey(turnChatId), + }) + } finally { + await db.delete(copilotChats).where(eq(copilotChats.id, turnChatId)) + } + } + + it("opens the workbench to a later turn while an earlier turn's workflow tool run goes on", async () => { + expect(await laterTurnWorkbenchAccess('run_workflow')).toEqual({ + handlersPending: false, + processes: [], + }) + }) + + it("keeps the workbench closed while an earlier turn's other tool is still executing", async () => { + expect(await laterTurnWorkbenchAccess('function_execute')).toMatchObject({ + handlersPending: true, + }) + }) }) diff --git a/apps/sim/lib/mothership/tools/client-executed-tools.ts b/apps/sim/lib/mothership/tools/client-executed-tools.ts index a0cc8af4382..8d4fe8555ca 100644 --- a/apps/sim/lib/mothership/tools/client-executed-tools.ts +++ b/apps/sim/lib/mothership/tools/client-executed-tools.ts @@ -1,14 +1,15 @@ import { isDesktopToolCall } from '@/lib/mothership/tools/desktop-tools' -const WORKFLOW_TOOL_NAMES = new Set([ +/** The tools that run a workflow, in the browser or on the server. */ +export const WORKFLOW_TOOL_NAMES = [ 'run_workflow', 'run_workflow_until_block', 'run_block', 'run_from_block', -]) +] as const export function isWorkflowToolName(name: string): boolean { - return WORKFLOW_TOOL_NAMES.has(name) + return (WORKFLOW_TOOL_NAMES as readonly string[]).includes(name) } /** diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts index e3bc642e2cd..80dcafcce04 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts @@ -262,6 +262,12 @@ function copilotRunLifecycle(context: ExecutionContext) { /** How often a run past its wait is re-read for its log before it is answered running. */ const RUN_LOG_POLL_MS = 1000 +/** + * How long past its wait a run's log is polled for. A run still unreadable by then (stuck in + * preflight or startup) is held until it settles, without more reads. + */ +const RUN_LOG_POLL_LIMIT_MS = 60_000 + /** Whether the run's log exists; a lookup that fails counts as not yet, so the call keeps waiting. */ async function runIsReadable(executionId: string, workflowId: string): Promise { try { @@ -292,21 +298,39 @@ async function answerWithinWait( const executionId = lifecycle.boundExecution?.executionId ?? generateId() const settled = run({ ...lifecycle, executionId }) retainSandboxExecution(settled) + const outcome: { + current?: { ok: true; result: ToolCallResult } | { ok: false; error: unknown } + } = {} + const finished = settled.then( + (result) => { + outcome.current = { ok: true, result } + }, + (error: unknown) => { + outcome.current = { ok: false, error } + } + ) const stopWaiting = new AbortController() - const waited = (ms: number) => interruptibleSleep(ms, stopWaiting.signal).then(() => null) + const waited = (ms: number) => interruptibleSleep(ms, stopWaiting.signal) try { - let result = await Promise.race([settled, waited(waitMs)]) + await Promise.race([finished, waited(waitMs)]) // Answered running only once the run is readable by its id; one still starting up (or in // preflight) is waited on until its log exists or it settles. - while (!result && !(await runIsReadable(executionId, workflowId))) { - result = await Promise.race([settled, waited(RUN_LOG_POLL_MS)]) + const pollUntil = Date.now() + RUN_LOG_POLL_LIMIT_MS + let readable = false + while (!outcome.current && !readable) { + if (Date.now() >= pollUntil) { + await finished + break + } + readable = await runIsReadable(executionId, workflowId) + if (!readable) await Promise.race([finished, waited(RUN_LOG_POLL_MS)]) } - // A run that settled while its log was being read answers with its result. - result ??= await Promise.race([settled, Promise.resolve(null)]) - if (result) return result } finally { stopWaiting.abort() } + // A run that settled while its log was being read answers with how it settled. + if (outcome.current?.ok === false) throw outcome.current.error + if (outcome.current) return outcome.current.result logger.info('Workflow tool run outlasted its wait; answering it still running', { toolCallId: context.toolCallId, workflowId, From b5d6485654138026ca7a3fc261cbbc3c266b4383 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 11:16:03 -0700 Subject: [PATCH 2/3] improvement(mothership): read a detached run's call as flat columns, and name the paused case Co-Authored-By: Claude Opus 5.5 --- .../mothership/async-runs/repository.test.ts | 37 ++++++++------- .../lib/mothership/async-runs/repository.ts | 46 ++++++++++--------- .../workflow-client-settlement.integration.ts | 9 +++- 3 files changed, 51 insertions(+), 41 deletions(-) diff --git a/apps/sim/lib/mothership/async-runs/repository.test.ts b/apps/sim/lib/mothership/async-runs/repository.test.ts index b31c12226c0..c2abba93430 100644 --- a/apps/sim/lib/mothership/async-runs/repository.test.ts +++ b/apps/sim/lib/mothership/async-runs/repository.test.ts @@ -461,10 +461,25 @@ describe('async tool repository single-row semantics', () => { }) describe('the workflow run a settled call left running', () => { + const stored = { + argv: null, + exitCode: null, + stdout: null, + status: null, + executionId: null, + } const cli = (argv: string[], exitCode: number, record: Record) => ({ + ...stored, executableName: 'sim_cli', - args: { request: { invocation: { kind: 'cli', argv } } }, - result: { exitCode, stdout: JSON.stringify(record), stderr: '' }, + argv, + exitCode: String(exitCode), + stdout: JSON.stringify(record), + }) + const tool = (executableName: string, status: string | null, executionId: string) => ({ + ...stored, + executableName, + status, + executionId, }) const running = { runId: 'run-1', workflowId: 'wf-1', status: 'running' } @@ -485,13 +500,7 @@ describe('the workflow run a settled call left running', () => { }) it('is a workflow tool answered still running', () => { - expect( - detachedWorkflowExecutionId({ - executableName: 'run_workflow', - args: { workflowId: 'wf-1' }, - result: { status: 'running', executionId: 'exec-1', workflowId: 'wf-1' }, - }) - ).toBe('exec-1') + expect(detachedWorkflowExecutionId(tool('run_workflow', 'running', 'exec-1'))).toBe('exec-1') }) it.each([ @@ -503,15 +512,9 @@ describe('the workflow run a settled call left running', () => { ['a wait on a run', cli(['workflows', 'runs', 'wait', 'run-1'], 4, running)], [ 'an unreadable answer', - { - ...cli(['workflows', 'run', 'wf-1'], 4, running), - result: { exitCode: 4, stdout: 'not json' }, - }, - ], - [ - 'a workflow tool that settled', - { executableName: 'run_block', args: {}, result: { success: true, executionId: 'exec-1' } }, + { ...cli(['workflows', 'run', 'wf-1'], 4, running), stdout: 'not json' }, ], + ['a workflow tool that settled', tool('run_block', null, 'exec-1')], ])('is not %s', (_label, call) => { expect(detachedWorkflowExecutionId(call)).toBeUndefined() }) diff --git a/apps/sim/lib/mothership/async-runs/repository.ts b/apps/sim/lib/mothership/async-runs/repository.ts index f2b236f4ad9..0b6a8c50894 100644 --- a/apps/sim/lib/mothership/async-runs/repository.ts +++ b/apps/sim/lib/mothership/async-runs/repository.ts @@ -1826,10 +1826,7 @@ export async function prepareWorkbenchAccess( if (processes.some((process) => process.sessionKey !== input.sessionKey)) throw new Error('Recorded workbench ownership does not match this chat') return { - /** - * A workflow tool's handler runs a workflow, which never touches the workbench. One answered - * still running holds its run past its turn, so it must not hold the workbench too. - */ + // A workflow tool's run never touches the workbench, and one answered running outlives its turn. handlersPending: previous.some( (row) => !row.settledAt && @@ -1940,30 +1937,34 @@ export async function getUnsettledClientWorkflowExecutions( ) } +/** The fields of a stored call that say whether it left a workflow run going, read as text. */ +interface DetachedRunCall { + executableName: string | null + /** `args.request.invocation.argv` of a `sim_cli` call. */ + argv: unknown + exitCode: string | null + stdout: string | null + status: string | null + executionId: string | null +} + /** * The workflow execution a settled call started and left running: a `workflows run` the CLI * answered still running (exit 4 and its run record), or a workflow tool answered still running * (`WorkflowToolRunningResult`). A run the call only waited on (`workflows runs wait`) is not * one: another run started it. */ -export function detachedWorkflowExecutionId(call: { - executableName: string | null - args: unknown - result: unknown -}): string | undefined { - const result = toRecordOrNull(call.result) - if (!result) return undefined +export function detachedWorkflowExecutionId(call: DetachedRunCall): string | undefined { if (call.executableName === 'sim_cli') { - const request = toRecordOrNull(toRecordOrNull(call.args)?.request) - const argv = toRecordOrNull(request?.invocation)?.argv + const argv = call.argv // The command path, wherever global flags (`--output json`) put it, never `workflows runs`. const end = Array.isArray(argv) && argv.includes('--') ? argv.indexOf('--') : undefined const tokens = Array.isArray(argv) ? argv.slice(0, end) : [] if (!tokens.some((token, index) => token === 'workflows' && tokens[index + 1] === 'run')) return undefined - if (result.exitCode !== 4 || typeof result.stdout !== 'string') return undefined + if (call.exitCode !== '4' || call.stdout === null) return undefined try { - const record = toRecordOrNull(JSON.parse(result.stdout)) + const record = toRecordOrNull(JSON.parse(call.stdout)) return record?.status === 'running' && typeof record.runId === 'string' ? record.runId : undefined @@ -1971,9 +1972,7 @@ export function detachedWorkflowExecutionId(call: { return undefined } } - return result.status === 'running' && typeof result.executionId === 'string' - ? result.executionId - : undefined + return call.status === 'running' && call.executionId !== null ? call.executionId : undefined } /** @@ -1995,12 +1994,15 @@ export async function getDetachedWorkflowExecutions( { [TraceAttr.StreamId]: streamId, [TraceAttr.UserId]: userId }, async () => { const { args, result } = copilotAsyncToolCalls - /** Only the fields `detachedWorkflowExecutionId` reads, so no call's payload is loaded. */ - const rows = await executor + // Only what `detachedWorkflowExecutionId` reads; an exit-4 `sim_cli` stdout is its run record. + const rows: DetachedRunCall[] = await executor .select({ executableName: copilotAsyncToolCalls.executableName, - args: sql`jsonb_build_object('request', jsonb_build_object('invocation', jsonb_build_object('argv', ${args} #> '{request,invocation,argv}')))`, - result: sql`jsonb_build_object('exitCode', ${result} -> 'exitCode', 'stdout', ${result} -> 'stdout', 'status', ${result} -> 'status', 'executionId', ${result} -> 'executionId')`, + argv: sql`${args} #> '{request,invocation,argv}'`, + exitCode: sql`${result} ->> 'exitCode'`, + stdout: sql`${result} ->> 'stdout'`, + status: sql`${result} ->> 'status'`, + executionId: sql`${result} ->> 'executionId'`, }) .from(copilotAsyncToolCalls) .innerJoin(copilotRuns, eq(copilotRuns.id, copilotAsyncToolCalls.runId)) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts b/apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts index a332e2fd007..5f35f02f68b 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-settlement.integration.ts @@ -591,9 +591,14 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => ) }) - it.each(['pending', 'completed', 'failed', 'cancelled'] as const)( + it.each([ + ['pending (paused at a human approval)', 'pending'], + ['completed', 'completed'], + ['failed', 'failed'], + ['cancelled', 'cancelled'], + ] as const)( 'leaves a detached run whose log is %s for a Stop to skip', - async (logStatus) => { + async (_label, logStatus) => { const detachedCli = generateId() const detachedTool = generateId() const cliCall = generateId() From 335ad5c17c907f393ab46f509a9203c72731f426 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 11:20:55 -0700 Subject: [PATCH 3/3] fix(mothership): answer a workflow tool run that settles as its log is read with its result Co-Authored-By: Claude Opus 5.5 --- .../tools/handlers/workflow/mutations.test.ts | 18 ++++++++++++++++++ .../tools/handlers/workflow/mutations.ts | 3 ++- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts index e1d3593109f..d06672353ef 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts @@ -476,6 +476,8 @@ describe('a workflow tool run past its wait', () => { executeRunWorkflow({ workflowId: 'workflow-1' }, { ...context, workflowRunWaitMs: 240_000 }) ) await vi.advanceTimersByTimeAsync(240_000) + // The one-macrotask settle check after the log read (a 0 ms timer runs at 1 ms). + await vi.advanceTimersByTimeAsync(1) const result = await answer const executionId = run.lifecycle().executionId expect(executionId).toEqual(expect.any(String)) @@ -525,11 +527,26 @@ describe('a workflow tool run past its wait', () => { await vi.advanceTimersByTimeAsync(1_000) expect(answered).toBe(false) await vi.advanceTimersByTimeAsync(1_000) + await vi.advanceTimersByTimeAsync(1) await expect(answer).resolves.toMatchObject({ output: { status: 'running' } }) expect(mocks.logStatus).toHaveBeenCalledWith(run.lifecycle().executionId, 'workflow-1') run.settle({ success: true, output: {}, logs: [] }) }) + it('answers a run that settles as its log read returns with how it settled', async () => { + const run = pendingRun() + mocks.logStatus.mockReset().mockImplementation(async () => { + queueMicrotask(() => run.settle({ success: true, output: { value: 5 }, logs: [] })) + return 'running' + }) + const answer = executeRunWorkflow( + { workflowId: 'workflow-1' }, + { ...context, workflowRunWaitMs: 1_000 } + ) + await vi.advanceTimersByTimeAsync(1_000) + await expect(answer).resolves.toMatchObject({ success: true, output: { output: { value: 5 } } }) + }) + it('keeps waiting on a run past its wait when its log cannot be read', async () => { const run = pendingRun() mocks.logStatus.mockReset().mockRejectedValue(new Error('db down')) @@ -574,6 +591,7 @@ describe('a workflow tool run past its wait', () => { { ...context, boundWorkflowExecutionId: 'bound-1', workflowRunWaitMs: 1_000 } ) await vi.advanceTimersByTimeAsync(1_000) + await vi.advanceTimersByTimeAsync(1) await expect(answer).resolves.toMatchObject({ output: { status: 'running', executionId: 'bound-1' }, }) diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts index 80dcafcce04..9f0b432b52a 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts @@ -325,10 +325,11 @@ async function answerWithinWait( readable = await runIsReadable(executionId, workflowId) if (!readable) await Promise.race([finished, waited(RUN_LOG_POLL_MS)]) } + // One macrotask lets a run that settled as its log was read deliver its result first. + if (!outcome.current) await Promise.race([finished, waited(0)]) } finally { stopWaiting.abort() } - // A run that settled while its log was being read answers with how it settled. if (outcome.current?.ok === false) throw outcome.current.error if (outcome.current) return outcome.current.result logger.info('Workflow tool run outlasted its wait; answering it still running', {