Skip to content

Commit 2e2f613

Browse files
committed
test(mothership): drive a draft run past a wait slice through the real controller
A live case for the worker's sliced synchronous `workflows run` (#9464, #9469): a draft with Wait blocks totalling 500 s runs through runCopilotLifecycle, the sim_cli handler, the embedded CLI and the v2 execute and runs routes against the worker's scripted draft-run probe. With workflow-run-wait-timeout declared it checks the detached first slice, the `workflows runs wait` slices, the output read-back, that no call holds the run and no slice row is failed, and that the model receives the run's output. Without the capability it checks one held call as before, and a model's own --wait-timeout 5 reaches the model as exit 4 with the running receipt. The controller fixture now pins direct Sim transport, since self-hosted defaults to checkpoint delivery and its run control went unanswered, and the fixture plan's execution timeout is per case so a run can outlast a slice.
1 parent d2f09ff commit 2e2f613

1 file changed

Lines changed: 346 additions & 2 deletions

File tree

‎apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts‎

Lines changed: 346 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,9 @@
11
/**
22
* Opt in with MSHIP_TEST_DATABASE_URL pointing to a local mship_audit_* database.
33
* SIM_HELPERS_SMOKE=1 also runs the actual service/DAG/Function runtime and log writer.
4-
* MSHIP_WORKER_ROOT additionally runs the real controller with a local scripted worker.
4+
* MSHIP_WORKER_ROOT additionally runs the real controller with a local scripted worker
5+
* (MSHIP_PROBE_REDIS_URL picks its Redis); the long draft-run cases take about nine minutes and
6+
* write their evidence to MSHIP_SLICES_REPORT_PATH.<capability>.json when it is set.
57
* MSHIP_LOCAL_COMPUTE_IMAGE opts into local Docker shell computation (Python 3 required).
68
* Those cases run the actual Function adapter and Mothership file/table post-processors;
79
* mounts, session directories and result collection replace the remote provider boundary.
@@ -130,6 +132,7 @@ import {
130132
} from '@/lib/mothership/chat/process-contents'
131133
import { finalizeAssistantTurn } from '@/lib/mothership/chat/terminal-state'
132134
import { createRegistryContentGate } from '@/lib/mothership/content-delivery/registry-gate'
135+
import { WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY } from '@/lib/mothership/generated/protocol'
133136
import { prepareInboxAttachments } from '@/lib/mothership/inbox/attachments'
134137
import { claimRunController } from '@/lib/mothership/request/lifecycle/controller-ownership'
135138
import { runCopilotLifecycle } from '@/lib/mothership/request/lifecycle/run'
@@ -194,6 +197,7 @@ const fixture = vi.hoisted(() => ({
194197
storageKeys: new Map<string, string>(),
195198
inboxBytes: new Map<string, Buffer>(),
196199
workbenchBootstrap: '',
200+
executionTimeoutMs: 20_000,
197201
}))
198202
vi.mock('@/lib/mothership/request/http', async (importOriginal) => ({
199203
...(await importOriginal<typeof import('@/lib/mothership/request/http')>()),
@@ -294,6 +298,15 @@ vi.mock('@/lib/mothership/server/agent-url', async (importOriginal) => ({
294298
...(await importOriginal<typeof import('@/lib/mothership/server/agent-url')>()),
295299
getMothershipBaseURL: () => fixture.workerUrl,
296300
}))
301+
// The control fixture answers run control on its direct endpoint; self-hosted defaults to checkpoint.
302+
vi.mock('@/lib/mothership/transport/connection', async (importOriginal) => {
303+
const actual = await importOriginal<typeof import('@/lib/mothership/transport/connection')>()
304+
return {
305+
...actual,
306+
getSimConnection: (override?: 'direct' | 'checkpoint') =>
307+
actual.getSimConnection(override ?? 'direct'),
308+
}
309+
})
297310
vi.mock('@/lib/mothership/chat/delegation', () => ({
298311
mintDelegationToken: async () => 'local-key',
299312
}))
@@ -335,7 +348,7 @@ vi.mock('@/lib/execution/preprocessing', async (importOriginal) => ({
335348
billingPeriod: { start: '2026-09-01T00:00:00.000Z', end: '2026-10-01T00:00:00.000Z' },
336349
payerSubscription: null,
337350
},
338-
executionTimeout: { sync: 20_000, async: 20_000 },
351+
executionTimeout: { sync: fixture.executionTimeoutMs, async: fixture.executionTimeoutMs },
339352
}),
340353
}))
341354
vi.mock('@/lib/internal/principals/executor', async (importOriginal) => ({
@@ -3580,6 +3593,337 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)(
35803593
40_000
35813594
)
35823595

3596+
it.skipIf(process.env.SIM_HELPERS_SMOKE !== '1' || !process.env.MSHIP_WORKER_ROOT).each([
3597+
// capability, seconds per Wait block (an in-process wait is capped at 300 s). The
3598+
// worker waits in 240 s slices: 500 s is a detached run, two run waits and a read-back.
3599+
['declared', 300, 200],
3600+
['undeclared', 260, 0],
3601+
// The model's own --wait-timeout 5 lapses: an exit 4 the worker hands the model.
3602+
['own-budget', 30, 0],
3603+
] as const)(
3604+
'draft test run past a wait slice reaches the model settled (capability %s)',
3605+
async (capability, waitA, waitB) => {
3606+
const workerRoot = process.env.MSHIP_WORKER_ROOT
3607+
if (!workerRoot) throw new Error('MSHIP_WORKER_ROOT must point to the worker checkout')
3608+
const draftId = generateId()
3609+
await db.insert(workflow).values({
3610+
id: draftId,
3611+
workspaceId,
3612+
userId: 'run-reader',
3613+
name: `Long draft ${capability}`,
3614+
lastSynced: now,
3615+
createdAt: now,
3616+
updatedAt: now,
3617+
isDeployed: false,
3618+
})
3619+
const finishId = generateId()
3620+
const blocks: Record<string, BlockState> = {
3621+
start: {
3622+
id: 'start',
3623+
type: 'start_trigger',
3624+
name: 'Start',
3625+
position: { x: 0, y: 0 },
3626+
enabled: true,
3627+
subBlocks: {},
3628+
outputs: {},
3629+
},
3630+
}
3631+
const edges: WorkflowState['edges'] = []
3632+
let previous = 'start'
3633+
for (const [index, seconds] of [waitA, waitB].entries()) {
3634+
if (seconds <= 0) continue
3635+
const id = `wait-${index}`
3636+
blocks[id] = {
3637+
id,
3638+
type: 'wait',
3639+
name: `Wait ${index + 1}`,
3640+
position: { x: 0, y: 100 * (index + 1) },
3641+
enabled: true,
3642+
subBlocks: {
3643+
timeValue: { id: 'timeValue', type: 'short-input', value: String(seconds) },
3644+
timeUnit: { id: 'timeUnit', type: 'dropdown', value: 'seconds' },
3645+
},
3646+
outputs: {},
3647+
}
3648+
edges.push({ id: `${previous}-${id}`, source: previous, target: id })
3649+
previous = id
3650+
}
3651+
blocks[finishId] = {
3652+
id: finishId,
3653+
type: 'function',
3654+
name: 'Finish',
3655+
position: { x: 0, y: 400 },
3656+
enabled: true,
3657+
subBlocks: {
3658+
code: {
3659+
id: 'code',
3660+
type: 'code',
3661+
value: 'return { invoice: 42, settled: "after-waits" };',
3662+
},
3663+
language: { id: 'language', type: 'dropdown', value: 'javascript' },
3664+
},
3665+
outputs: { result: { type: 'json' } },
3666+
}
3667+
edges.push({ id: `${previous}-finish`, source: previous, target: finishId })
3668+
fixture.permission = 'write'
3669+
fixture.saved.set(draftId, { blocks, edges, loops: {}, parallels: {} })
3670+
// A paid plan's synchronous limit, so the run outlasts several slices.
3671+
fixture.executionTimeoutMs = 50 * 60_000
3672+
3673+
const child = spawn('bun', ['tools/probes/src/draft-run-slices.ts'], {
3674+
cwd: workerRoot,
3675+
env: {
3676+
PATH: process.env.PATH,
3677+
NODE_ENV: 'test',
3678+
DATABASE_URL: process.env.MSHIP_TEST_DATABASE_URL,
3679+
REDIS_URL: process.env.MSHIP_PROBE_REDIS_URL ?? 'redis://127.0.0.1:6379/14',
3680+
ANTHROPIC_API_KEY: 'local-scripted-provider',
3681+
COPILOT_INBOUND_API_KEY: 'local-controller-probe-key',
3682+
SIM_ENDPOINT: controlEndpoint,
3683+
INTERNAL_API_SECRET: env.INTERNAL_API_SECRET,
3684+
MSHIP_PROBE_WORKFLOW_ID: draftId,
3685+
MSHIP_PROBE_RUN_FLAGS: JSON.stringify(
3686+
capability === 'own-budget' ? ['--wait-timeout', '5'] : []
3687+
),
3688+
},
3689+
stdio: ['pipe', 'pipe', 'pipe'],
3690+
})
3691+
const exited = once(child, 'exit')
3692+
const lines = createInterface({ input: child.stdout })
3693+
let diagnostics = ''
3694+
child.stderr.on('data', (chunk: Buffer) => {
3695+
diagnostics += chunk.toString()
3696+
})
3697+
const probeEvents: Array<{ probe: string; at: number }> = []
3698+
const ready = new Promise<string>((resolve) => {
3699+
lines.on('line', (line) => {
3700+
diagnostics += `${line}\n`
3701+
if (!line.startsWith('{"probe":')) return
3702+
const message = JSON.parse(line)
3703+
if (typeof message.at === 'number') probeEvents.push(message)
3704+
if (typeof message.port === 'number') resolve(`http://127.0.0.1:${message.port}`)
3705+
})
3706+
})
3707+
const workerRequests: Array<{ path: string; at: number }> = []
3708+
try {
3709+
fixture.workerUrl = await Promise.race([
3710+
ready,
3711+
exited.then(() => {
3712+
throw new Error(`Worker probe exited before readiness: ${diagnostics}`)
3713+
}),
3714+
])
3715+
vi.stubEnv('COPILOT_API_KEY', 'local-controller-probe-key')
3716+
vi.unstubAllGlobals()
3717+
const nativeFetch = globalThis.fetch
3718+
vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => {
3719+
const url = new URL(input instanceof Request ? input.url : input)
3720+
if (url.origin !== fixture.workerUrl)
3721+
throw new Error(`Unexpected external request: ${url.origin}`)
3722+
workerRequests.push({ path: url.pathname, at: Date.now() })
3723+
return nativeFetch(input, init)
3724+
})
3725+
ensureHandlersRegistered()
3726+
const chatId = generateId()
3727+
const messageId = generateId()
3728+
await db.insert(copilotChats).values({
3729+
id: chatId,
3730+
userId: 'run-reader',
3731+
workspaceId,
3732+
type: 'mothership',
3733+
title: `Long draft ${capability}`,
3734+
})
3735+
const startedAt = Date.now()
3736+
const result = await runCopilotLifecycle(
3737+
{
3738+
message: 'Test-run the draft and tell me its output.',
3739+
workflowId: draftId,
3740+
userId: 'run-reader',
3741+
workspaceId,
3742+
chatId,
3743+
messageId,
3744+
// What lib/mothership/chat/payload.ts declares on every chat turn.
3745+
...(capability !== 'undeclared'
3746+
? { clientCapabilities: [WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY] }
3747+
: {}),
3748+
},
3749+
{
3750+
userId: 'run-reader',
3751+
workspaceId,
3752+
workflowId: draftId,
3753+
chatId,
3754+
goRoute: '/api/mothership',
3755+
interactive: false,
3756+
clientToolPickupExpected: false,
3757+
flushAfterEvent: false,
3758+
abortSignal: AbortSignal.timeout(20 * 60_000),
3759+
executionContext: {
3760+
userId: 'run-reader',
3761+
workspaceId,
3762+
workflowId: draftId,
3763+
chatId,
3764+
userPermission: 'write',
3765+
resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry([], {
3766+
userId: 'run-reader',
3767+
workspaceId,
3768+
}),
3769+
},
3770+
}
3771+
)
3772+
const endedAt = Date.now()
3773+
const [run] = await db
3774+
.select()
3775+
.from(copilotRuns)
3776+
.where(eq(copilotRuns.streamId, messageId))
3777+
const calls = run
3778+
? await db
3779+
.select()
3780+
.from(copilotAsyncToolCalls)
3781+
.where(eq(copilotAsyncToolCalls.runId, run.id))
3782+
: []
3783+
const evidence = {
3784+
capability,
3785+
waits: [waitA, waitB],
3786+
startedAt: new Date(startedAt).toISOString(),
3787+
totalSeconds: (endedAt - startedAt) / 1000,
3788+
success: result.success,
3789+
content: result.content,
3790+
runStatus: run?.status,
3791+
toolCalls: calls
3792+
.sort((a, b) => Number(a.createdAt) - Number(b.createdAt))
3793+
.map((call) => ({
3794+
toolCallId: call.toolCallId,
3795+
toolName: call.toolName,
3796+
args: call.args,
3797+
status: call.status,
3798+
exitCode: (call.result as { exitCode?: unknown } | null)?.exitCode,
3799+
stdout: (call.result as { stdout?: unknown } | null)?.stdout,
3800+
startedAt: call.executionStartedAt?.toISOString(),
3801+
settledAt: call.executionSettledAt?.toISOString(),
3802+
seconds:
3803+
call.executionStartedAt && call.executionSettledAt
3804+
? (Number(call.executionSettledAt) - Number(call.executionStartedAt)) / 1000
3805+
: undefined,
3806+
})),
3807+
workerRequests: workerRequests.map((request) => ({
3808+
...request,
3809+
at: new Date(request.at).toISOString(),
3810+
})),
3811+
probeEvents,
3812+
}
3813+
if (process.env.MSHIP_SLICES_REPORT_PATH)
3814+
await writeFile(
3815+
`${process.env.MSHIP_SLICES_REPORT_PATH}.${capability}.json`,
3816+
JSON.stringify(evidence, null, 2)
3817+
)
3818+
expect(
3819+
result.success,
3820+
JSON.stringify({ evidence, errors: fixture.errors, diagnostics })
3821+
).toBe(true)
3822+
expect(run?.status).toBe('complete')
3823+
const report = JSON.parse(result.content)
3824+
const seen = String(report.output.value)
3825+
if (capability === 'own-budget') {
3826+
expect(calls).toHaveLength(1)
3827+
expect(
3828+
(calls[0].args as { request: { invocation: { argv: string[] } } }).request.invocation
3829+
.argv
3830+
).toEqual(['workflows', 'run', draftId, '--manual', '--wait-timeout', '5'])
3831+
expect(calls[0].status).not.toBe('failed')
3832+
expect(seen).toContain('"exitCode":4')
3833+
expect(seen).toMatch(/"status": ?"running"/)
3834+
expect(seen).toContain('sim workflows runs wait')
3835+
await new Promise((resolve) => setTimeout(resolve, 35_000))
3836+
return
3837+
}
3838+
// Sim #9469: a still-running (exit 4) slice settles as done, not failed.
3839+
for (const call of calls) expect(call.status, call.toolName).not.toBe('failed')
3840+
// The model saw the settled run, not a detached receipt.
3841+
expect(seen).toContain('"exitCode":0')
3842+
expect(seen).toContain('"status":"completed"')
3843+
expect(seen).toContain('after-waits')
3844+
const waitSeconds = waitA + waitB
3845+
const argvOf = (call: (typeof calls)[number]) =>
3846+
(call.args as { request: { invocation: { argv: string[] } } }).request.invocation.argv
3847+
const outcome = (call: (typeof calls)[number]) =>
3848+
call.result as { exitCode: number; stdout: string }
3849+
const heldSeconds = (call: (typeof calls)[number]) =>
3850+
(Number(call.executionSettledAt) - Number(call.executionStartedAt)) / 1000
3851+
const runs = calls.filter((call) => argvOf(call)[1] === 'run')
3852+
const waits = calls.filter(
3853+
(call) => argvOf(call).slice(0, 3).join(' ') === 'workflows runs wait'
3854+
)
3855+
const gets = calls.filter(
3856+
(call) => argvOf(call).slice(0, 3).join(' ') === 'workflows runs get'
3857+
)
3858+
expect(runs).toHaveLength(1)
3859+
expect(calls).toHaveLength(runs.length + waits.length + gets.length)
3860+
if (capability === 'declared') {
3861+
// First slice: the run itself, bounded at one slice and answered still running.
3862+
expect(argvOf(runs[0])).toEqual([
3863+
'workflows',
3864+
'run',
3865+
draftId,
3866+
'--manual',
3867+
'--wait-timeout',
3868+
'240',
3869+
])
3870+
expect(outcome(runs[0]).exitCode).toBe(4)
3871+
const receipt = JSON.parse(outcome(runs[0]).stdout)
3872+
expect(receipt).toMatchObject({ workflowId: draftId, status: 'running' })
3873+
const runId = receipt.runId
3874+
expect(waits).toHaveLength(Math.ceil(waitSeconds / 240) - 1)
3875+
for (const [index, call] of waits.entries()) {
3876+
expect(argvOf(call)).toEqual([
3877+
'workflows',
3878+
'runs',
3879+
'wait',
3880+
runId,
3881+
'--workflow',
3882+
draftId,
3883+
'--wait-timeout',
3884+
'240',
3885+
])
3886+
expect(outcome(call).exitCode).toBe(index === waits.length - 1 ? 0 : 4)
3887+
}
3888+
expect(JSON.parse(outcome(waits.at(-1)!).stdout)).toMatchObject({
3889+
runId,
3890+
status: 'completed',
3891+
})
3892+
// Settled: read back once with its output, as the held call printed it.
3893+
expect(gets).toHaveLength(1)
3894+
expect(argvOf(gets[0])).toEqual([
3895+
'workflows',
3896+
'runs',
3897+
'get',
3898+
runId,
3899+
'--workflow',
3900+
draftId,
3901+
'--include-output',
3902+
])
3903+
expect(outcome(gets[0]).stdout).toContain('after-waits')
3904+
// No single Sim call held the whole run.
3905+
for (const call of calls) expect(heldSeconds(call)).toBeLessThan(250)
3906+
} else {
3907+
expect(argvOf(runs[0])).toEqual(['workflows', 'run', draftId, '--manual'])
3908+
expect(waits).toHaveLength(0)
3909+
expect(gets).toHaveLength(0)
3910+
expect(outcome(runs[0]).exitCode).toBe(0)
3911+
// One held call for the whole run, as on staging.
3912+
expect(heldSeconds(runs[0])).toBeGreaterThan(waitSeconds)
3913+
}
3914+
expect(endedAt - startedAt).toBeGreaterThan(waitSeconds * 1000)
3915+
} finally {
3916+
fixture.executionTimeoutMs = 20_000
3917+
child.kill()
3918+
await exited
3919+
lines.close()
3920+
vi.unstubAllGlobals()
3921+
vi.unstubAllEnvs()
3922+
}
3923+
},
3924+
25 * 60_000
3925+
)
3926+
35833927
it
35843928
.skipIf(process.env.SIM_HELPERS_SMOKE !== '1' || !process.env.MSHIP_WORKER_ROOT)
35853929
.each(['valid', 'rewrite'] as const)(

0 commit comments

Comments
 (0)