Skip to content

Commit 2306df4

Browse files
waleedlatif1waleedlatif1
committed
fix(mothership): stop the workflow runs a stopped chat left running past a call
A synchronous workflows run the worker waits on in slices detaches after its first --wait-timeout (#9464, mothership#872), and from then on no call's abort signal reaches it. Stop delivery cancelled only browser workflow runs bound to open calls, so a Stop mid-slice ended the chat while the draft run went on to completion. Driven through the real controller and the Stop route (live case in the next commit), the run was still running a minute after the Stop. Stop delivery now also cancels the runs this chat run's calls started and left running: a workflows run the CLI answered still running (exit 4 and its run record), and a workflow tool answered still running. A run the chat only waited on (workflows runs wait) is not one, since something else started it. The cancel is the same in-process abort and durable cancel flag a browser run gets; a detached run registers its in-process aborter. Autopsy: the slice tests on both sides covered waits that settle, lapse or lose an answer, and none stopped the chat mid-slice; the Stop tests seeded only open browser-run calls. stop-delivery.test.ts and the live declared-stop case now cover it.
1 parent a1748e5 commit 2306df4

6 files changed

Lines changed: 256 additions & 28 deletions

File tree

‎apps/sim/lib/mothership/async-runs/repository.test.ts‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@ import {
1414
closeStreamToolAdmission,
1515
createRunSegment,
1616
detachAsyncToolCall,
17+
detachedWorkflowExecutionId,
1718
getClaimedWorkflowExecutionId,
1819
getUnsettledStreamSandboxProcesses,
1920
markAsyncToolRunning,
@@ -458,3 +459,52 @@ describe('async tool repository single-row semantics', () => {
458459
).rejects.toThrow('Async tool call belongs to another run')
459460
})
460461
})
462+
463+
describe('the workflow run a settled call left running', () => {
464+
const cli = (argv: string[], exitCode: number, record: Record<string, unknown>) => ({
465+
executableName: 'sim_cli',
466+
args: { request: { invocation: { kind: 'cli', argv } } },
467+
result: { exitCode, stdout: JSON.stringify(record), stderr: '' },
468+
})
469+
const running = { runId: 'run-1', workflowId: 'wf-1', status: 'running' }
470+
471+
it('is a workflows run the CLI answered still running', () => {
472+
expect(
473+
detachedWorkflowExecutionId(
474+
cli(['workflows', 'run', 'wf-1', '--wait-timeout', '240'], 4, running)
475+
)
476+
).toBe('run-1')
477+
})
478+
479+
it('is a workflow tool answered still running', () => {
480+
expect(
481+
detachedWorkflowExecutionId({
482+
executableName: 'run_workflow',
483+
args: { workflowId: 'wf-1' },
484+
result: { status: 'running', executionId: 'exec-1', workflowId: 'wf-1' },
485+
})
486+
).toBe('exec-1')
487+
})
488+
489+
it.each([
490+
[
491+
'a run that settled',
492+
cli(['workflows', 'run', 'wf-1'], 0, { ...running, status: 'completed' }),
493+
],
494+
// Waiting on a run proves nothing about who started it; Stop must not cancel it.
495+
['a wait on a run', cli(['workflows', 'runs', 'wait', 'run-1'], 4, running)],
496+
[
497+
'an unreadable answer',
498+
{
499+
...cli(['workflows', 'run', 'wf-1'], 4, running),
500+
result: { exitCode: 4, stdout: 'not json' },
501+
},
502+
],
503+
[
504+
'a workflow tool that settled',
505+
{ executableName: 'run_block', args: {}, result: { success: true, executionId: 'exec-1' } },
506+
],
507+
])('is not %s', (_label, call) => {
508+
expect(detachedWorkflowExecutionId(call)).toBeUndefined()
509+
})
510+
})

‎apps/sim/lib/mothership/async-runs/repository.ts‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -261,6 +261,8 @@ export interface RunStopTarget {
261261
/** The client workflows and sandbox commands a stopped run started, read once its admission closed. */
262262
export interface StoppedRunWork {
263263
clientWorkflowExecutionIds: string[]
264+
/** Runs its calls started and left running past the call (`getDetachedWorkflowExecutions`). */
265+
detachedWorkflowExecutionIds: string[]
264266
sandboxProcesses: StoppedSandboxProcess[]
265267
}
266268

@@ -361,6 +363,7 @@ export async function stopLiveRunsTx(
361363
chatId,
362364
desktopDeviceId,
363365
clientWorkflowExecutionIds: await getUnsettledClientWorkflowExecutions(streamId, userId, tx),
366+
detachedWorkflowExecutionIds: await getDetachedWorkflowExecutions(streamId, userId, tx),
364367
sandboxProcesses: await getUnsettledStreamSandboxProcesses(streamId, userId, tx),
365368
})
366369
}
@@ -1922,6 +1925,81 @@ export async function getUnsettledClientWorkflowExecutions(
19221925
)
19231926
}
19241927

1928+
const WORKFLOW_RUN_TOOL_NAMES = [
1929+
'run_workflow',
1930+
'run_workflow_until_block',
1931+
'run_from_block',
1932+
'run_block',
1933+
] as const
1934+
1935+
/**
1936+
* The workflow execution a settled call started and left running: a `workflows run` the CLI
1937+
* answered still running (exit 4 and its run record), or a workflow tool answered still running
1938+
* (`WorkflowToolRunningResult`). A run the call only waited on (`workflows runs wait`) is not
1939+
* one: another run started it.
1940+
*/
1941+
export function detachedWorkflowExecutionId(call: {
1942+
executableName: string | null
1943+
args: unknown
1944+
result: unknown
1945+
}): string | undefined {
1946+
const result = toRecordOrNull(call.result)
1947+
if (!result) return undefined
1948+
if (call.executableName === 'sim_cli') {
1949+
const request = toRecordOrNull(toRecordOrNull(call.args)?.request)
1950+
const argv = toRecordOrNull(request?.invocation)?.argv
1951+
if (!Array.isArray(argv) || argv[0] !== 'workflows' || argv[1] !== 'run') return undefined
1952+
if (result.exitCode !== 4 || typeof result.stdout !== 'string') return undefined
1953+
try {
1954+
const record = toRecordOrNull(JSON.parse(result.stdout))
1955+
return record?.status === 'running' && typeof record.runId === 'string'
1956+
? record.runId
1957+
: undefined
1958+
} catch {
1959+
return undefined
1960+
}
1961+
}
1962+
return result.status === 'running' && typeof result.executionId === 'string'
1963+
? result.executionId
1964+
: undefined
1965+
}
1966+
1967+
/**
1968+
* Workflow executions this run's calls started and left running past the call: what a Stop
1969+
* cancels beyond the calls still open, since a detached run takes no call's abort signal.
1970+
*/
1971+
export async function getDetachedWorkflowExecutions(
1972+
streamId: string,
1973+
userId: string,
1974+
executor: DbOrTx = db
1975+
) {
1976+
return await withDbSpan(
1977+
TraceSpan.CopilotAsyncRunsGetRunSegment,
1978+
'SELECT',
1979+
'copilot_async_tool_calls',
1980+
{ [TraceAttr.StreamId]: streamId, [TraceAttr.UserId]: userId },
1981+
async () => {
1982+
const rows = await executor
1983+
.select({
1984+
executableName: copilotAsyncToolCalls.executableName,
1985+
args: copilotAsyncToolCalls.args,
1986+
result: copilotAsyncToolCalls.result,
1987+
})
1988+
.from(copilotAsyncToolCalls)
1989+
.innerJoin(copilotRuns, eq(copilotRuns.id, copilotAsyncToolCalls.runId))
1990+
.where(
1991+
and(
1992+
eq(copilotRuns.streamId, streamId),
1993+
eq(copilotRuns.userId, userId),
1994+
inArray(copilotAsyncToolCalls.executableName, ['sim_cli', ...WORKFLOW_RUN_TOOL_NAMES]),
1995+
isNotNull(copilotAsyncToolCalls.result)
1996+
)
1997+
)
1998+
return [...new Set(rows.flatMap((row) => detachedWorkflowExecutionId(row) ?? []))]
1999+
}
2000+
)
2001+
}
2002+
19252003
export function getClaimedWorkflowExecutionId(claimedBy: string | null | undefined) {
19262004
if (!claimedBy?.startsWith(WORKFLOW_EXECUTION_CLAIM_PREFIX)) return undefined
19272005
const executionId = claimedBy.slice(WORKFLOW_EXECUTION_CLAIM_PREFIX.length)

‎apps/sim/lib/mothership/request/application/stop-delivery.test.ts‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
import {
2+
mothershipAsyncRunsMock,
3+
mothershipAsyncRunsMockFns,
4+
} from '@sim/testing/mocks/mothership-async-runs.mock'
5+
import { beforeEach, describe, expect, it, vi } from 'vitest'
6+
7+
const hoisted = vi.hoisted(() => ({
8+
abortManualExecution: vi.fn(),
9+
markExecutionCancelled: vi.fn(),
10+
}))
11+
vi.mock('@/lib/mothership/async-runs/repository', () => mothershipAsyncRunsMock)
12+
vi.mock('@/lib/execution/manual-cancellation', () => ({
13+
abortManualExecution: hoisted.abortManualExecution,
14+
}))
15+
vi.mock('@/lib/execution/cancellation', () => ({
16+
markExecutionCancelled: hoisted.markExecutionCancelled,
17+
}))
18+
vi.mock('@/lib/mothership/request/session/explicit-abort', () => ({
19+
requestExplicitStreamAbort: vi.fn(async () => ({ settled: true })),
20+
}))
21+
vi.mock('@/lib/mothership/request/session', () => ({
22+
abortActiveStream: vi.fn(),
23+
waitForPendingChatStream: vi.fn(async () => true),
24+
releasePendingChatStream: vi.fn(),
25+
}))
26+
vi.mock('@/lib/mothership/async-runs/orphaned-runs', () => ({
27+
settleStoppedRunWithoutController: vi.fn(async () => undefined),
28+
}))
29+
vi.mock('@/lib/mothership/request/tools/sandbox-recovery', () => ({
30+
recoverSandboxProcesses: vi.fn(async () => undefined),
31+
}))
32+
vi.mock('@/lib/desktop/executor/doorbell', () => ({ ringDesktopInbox: vi.fn() }))
33+
34+
import {
35+
deliverRunStop,
36+
readStoppedRunWork,
37+
} from '@/lib/mothership/request/application/stop-delivery'
38+
39+
const run = { id: 'run-1', streamId: 'stream-1', userId: 'user-1', chatId: 'chat-1' }
40+
41+
describe('a Stop', () => {
42+
beforeEach(() => {
43+
hoisted.abortManualExecution.mockReset()
44+
hoisted.markExecutionCancelled.mockReset().mockResolvedValue({ durablyRecorded: true })
45+
})
46+
47+
// A synchronous run the worker waits on in slices has detached from its call: no call's abort
48+
// signal reaches it, so before this a Stop mid-slice left the draft running to completion.
49+
it('cancels the workflow runs its calls started and left running, beside its open client runs', async () => {
50+
mothershipAsyncRunsMockFns.mockGetUnsettledClientWorkflowExecutions.mockResolvedValueOnce([
51+
'client-1',
52+
])
53+
mothershipAsyncRunsMockFns.mockGetDetachedWorkflowExecutions.mockResolvedValueOnce([
54+
'detached-1',
55+
'client-1',
56+
])
57+
const work = await readStoppedRunWork('stream-1', 'user-1')
58+
expect(work).toEqual({
59+
clientWorkflowExecutionIds: ['client-1'],
60+
detachedWorkflowExecutionIds: ['detached-1', 'client-1'],
61+
sandboxProcesses: [],
62+
})
63+
64+
await deliverRunStop(run, work)
65+
66+
expect(hoisted.abortManualExecution.mock.calls).toEqual([['client-1'], ['detached-1']])
67+
expect(hoisted.markExecutionCancelled.mock.calls).toEqual([['client-1'], ['detached-1']])
68+
})
69+
70+
it('still cancels what it found when the detached runs cannot be read', async () => {
71+
mothershipAsyncRunsMockFns.mockGetUnsettledClientWorkflowExecutions.mockResolvedValueOnce([
72+
'client-1',
73+
])
74+
mothershipAsyncRunsMockFns.mockGetDetachedWorkflowExecutions.mockRejectedValueOnce(
75+
new Error('db down')
76+
)
77+
await deliverRunStop(run, await readStoppedRunWork('stream-1', 'user-1'))
78+
expect(hoisted.markExecutionCancelled.mock.calls).toEqual([['client-1']])
79+
})
80+
})

‎apps/sim/lib/mothership/request/application/stop-delivery.ts‎

Lines changed: 43 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import { markExecutionCancelled } from '@/lib/execution/cancellation'
55
import { abortManualExecution } from '@/lib/execution/manual-cancellation'
66
import { settleStoppedRunWithoutController } from '@/lib/mothership/async-runs/orphaned-runs'
77
import {
8+
getDetachedWorkflowExecutions,
89
getUnsettledClientWorkflowExecutions,
910
getUnsettledStreamSandboxProcesses,
1011
type RunStopTarget,
@@ -34,30 +35,39 @@ interface RunStopDeliveryOptions {
3435
}
3536

3637
/**
37-
* The client workflows and sandbox commands a stopped run started, read once its admission
38-
* closed. Each read is best-effort: one that fails stops nothing it would have found.
38+
* The client workflows, detached workflow runs and sandbox commands a stopped run started, read
39+
* once its admission closed. Each read is best-effort: one that fails stops nothing it would have
40+
* found.
3941
*/
4042
export async function readStoppedRunWork(
4143
streamId: string,
4244
userId: string
4345
): Promise<StoppedRunWork> {
44-
const [clientWorkflowExecutionIds, sandboxProcesses] = await Promise.all([
45-
getUnsettledClientWorkflowExecutions(streamId, userId).catch((error) => {
46-
logger.warn('Stopped stream workflow ownership could not be read', {
47-
streamId,
48-
error: getErrorMessage(error),
49-
})
50-
return []
51-
}),
52-
getUnsettledStreamSandboxProcesses(streamId, userId).catch((error) => {
53-
logger.warn('Stopped stream sandbox ownership could not be read', {
54-
streamId,
55-
error: getErrorMessage(error),
56-
})
57-
return []
58-
}),
59-
])
60-
return { clientWorkflowExecutionIds, sandboxProcesses }
46+
const [clientWorkflowExecutionIds, detachedWorkflowExecutionIds, sandboxProcesses] =
47+
await Promise.all([
48+
getUnsettledClientWorkflowExecutions(streamId, userId).catch((error) => {
49+
logger.warn('Stopped stream workflow ownership could not be read', {
50+
streamId,
51+
error: getErrorMessage(error),
52+
})
53+
return []
54+
}),
55+
getDetachedWorkflowExecutions(streamId, userId).catch((error) => {
56+
logger.warn('Stopped stream detached workflow runs could not be read', {
57+
streamId,
58+
error: getErrorMessage(error),
59+
})
60+
return []
61+
}),
62+
getUnsettledStreamSandboxProcesses(streamId, userId).catch((error) => {
63+
logger.warn('Stopped stream sandbox ownership could not be read', {
64+
streamId,
65+
error: getErrorMessage(error),
66+
})
67+
return []
68+
}),
69+
])
70+
return { clientWorkflowExecutionIds, detachedWorkflowExecutionIds, sandboxProcesses }
6171
}
6272

6373
/**
@@ -97,14 +107,21 @@ export async function deliverRunStop(
97107
error: getErrorMessage(error),
98108
})
99109
}
100-
const cancelClientWorkflows = async () =>
101-
Promise.all(
102-
(await work).clientWorkflowExecutionIds.map(async (executionId) => {
103-
/** The execution route reserved these IDs before recording ownership; no client result is trusted. */
104-
abortManualExecution(executionId)
105-
await markExecutionCancelled(executionId)
106-
})
110+
const cancelClientWorkflows = async () => {
111+
const { clientWorkflowExecutionIds, detachedWorkflowExecutionIds } = await work
112+
/**
113+
* The execution route reserved the client IDs before recording ownership; no client result is
114+
* trusted. A detached run took no call's abort signal, so only its own cancel reaches it.
115+
*/
116+
return Promise.all(
117+
[...new Set([...clientWorkflowExecutionIds, ...detachedWorkflowExecutionIds])].map(
118+
async (executionId) => {
119+
abortManualExecution(executionId)
120+
await markExecutionCancelled(executionId)
121+
}
122+
)
107123
)
124+
}
108125
const [settled, worker] = await Promise.all([
109126
waitForPendingChatStream(chatId, 8000, streamId),
110127
workerStop,

‎apps/sim/lib/organizations/member-manager.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -107,6 +107,7 @@ describe('organization member managers', () => {
107107
chatId: 'chat',
108108
desktopDeviceId: null,
109109
clientWorkflowExecutionIds: ['execution'],
110+
detachedWorkflowExecutionIds: [],
110111
sandboxProcesses: [],
111112
},
112113
]

‎packages/testing/src/mocks/mothership-async-runs.mock.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,8 @@ import { vi } from 'vitest'
88
* its result (the advisory lock itself is not modelled).
99
* - `mockGetRunSecretAuthority` resolves the no-secrets authority.
1010
* - `mockGetClaimedWorkflowExecutionId` is a faithful port (strips the real `'workflow:'` claim prefix).
11-
* - `mockStopLiveRunsTx`, `mockFindWorkspacesWithLiveRunsTx`, `mockGetUnsettledStreamSandboxProcesses`, `mockGetUnsettledClientWorkflowExecutions` and
12-
* `mockGetAsyncToolCalls` resolve `[]`.
11+
* - `mockStopLiveRunsTx`, `mockFindWorkspacesWithLiveRunsTx`, `mockGetUnsettledStreamSandboxProcesses`, `mockGetUnsettledClientWorkflowExecutions`,
12+
* `mockGetDetachedWorkflowExecutions` and `mockGetAsyncToolCalls` resolve `[]`.
1313
*
1414
* @example
1515
* ```ts
@@ -64,6 +64,7 @@ export const mothershipAsyncRunsMockFns = {
6464
mockGetUnsettledClientWorkflowExecutions: vi.fn(
6565
async (..._args: unknown[]): Promise<unknown[]> => []
6666
),
67+
mockGetDetachedWorkflowExecutions: vi.fn(async (..._args: unknown[]): Promise<unknown[]> => []),
6768
mockGetClaimedWorkflowExecutionId: vi.fn(
6869
(claimedBy: string | null | undefined): string | undefined => {
6970
if (!claimedBy?.startsWith('workflow:')) return undefined
@@ -131,6 +132,7 @@ export const mothershipAsyncRunsMock = {
131132
areStreamToolExecutionsSettled: mothershipAsyncRunsMockFns.mockAreStreamToolExecutionsSettled,
132133
getUnsettledClientWorkflowExecutions:
133134
mothershipAsyncRunsMockFns.mockGetUnsettledClientWorkflowExecutions,
135+
getDetachedWorkflowExecutions: mothershipAsyncRunsMockFns.mockGetDetachedWorkflowExecutions,
134136
getClaimedWorkflowExecutionId: mothershipAsyncRunsMockFns.mockGetClaimedWorkflowExecutionId,
135137
claimWorkflowToolExecution: mothershipAsyncRunsMockFns.mockClaimWorkflowToolExecution,
136138
releaseWorkflowToolExecutionClaim:

0 commit comments

Comments
 (0)