From 2e2f613fe9698f1c624bf616d66f3b7683b056cd Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 30 Oct 2026 10:04:06 -0700 Subject: [PATCH 1/2] 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` (simstudioai/sim#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. --- .../agent-cli/saved-run-read.live.test.ts | 348 +++++++++++++++++- 1 file changed, 346 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts b/apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts index ab21b3fbed4..9ab6bdfc9ad 100644 --- a/apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts +++ b/apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts @@ -1,7 +1,9 @@ /** * Opt in with MSHIP_TEST_DATABASE_URL pointing to a local mship_audit_* database. * SIM_HELPERS_SMOKE=1 also runs the actual service/DAG/Function runtime and log writer. - * MSHIP_WORKER_ROOT additionally runs the real controller with a local scripted worker. + * MSHIP_WORKER_ROOT additionally runs the real controller with a local scripted worker + * (MSHIP_PROBE_REDIS_URL picks its Redis); the long draft-run cases take about nine minutes and + * write their evidence to MSHIP_SLICES_REPORT_PATH..json when it is set. * MSHIP_LOCAL_COMPUTE_IMAGE opts into local Docker shell computation (Python 3 required). * Those cases run the actual Function adapter and Mothership file/table post-processors; * mounts, session directories and result collection replace the remote provider boundary. @@ -130,6 +132,7 @@ import { } from '@/lib/mothership/chat/process-contents' import { finalizeAssistantTurn } from '@/lib/mothership/chat/terminal-state' import { createRegistryContentGate } from '@/lib/mothership/content-delivery/registry-gate' +import { WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY } from '@/lib/mothership/generated/protocol' import { prepareInboxAttachments } from '@/lib/mothership/inbox/attachments' import { claimRunController } from '@/lib/mothership/request/lifecycle/controller-ownership' import { runCopilotLifecycle } from '@/lib/mothership/request/lifecycle/run' @@ -194,6 +197,7 @@ const fixture = vi.hoisted(() => ({ storageKeys: new Map(), inboxBytes: new Map(), workbenchBootstrap: '', + executionTimeoutMs: 20_000, })) vi.mock('@/lib/mothership/request/http', async (importOriginal) => ({ ...(await importOriginal()), @@ -294,6 +298,15 @@ vi.mock('@/lib/mothership/server/agent-url', async (importOriginal) => ({ ...(await importOriginal()), getMothershipBaseURL: () => fixture.workerUrl, })) +// The control fixture answers run control on its direct endpoint; self-hosted defaults to checkpoint. +vi.mock('@/lib/mothership/transport/connection', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + getSimConnection: (override?: 'direct' | 'checkpoint') => + actual.getSimConnection(override ?? 'direct'), + } +}) vi.mock('@/lib/mothership/chat/delegation', () => ({ mintDelegationToken: async () => 'local-key', })) @@ -335,7 +348,7 @@ vi.mock('@/lib/execution/preprocessing', async (importOriginal) => ({ billingPeriod: { start: '2026-09-01T00:00:00.000Z', end: '2026-10-01T00:00:00.000Z' }, payerSubscription: null, }, - executionTimeout: { sync: 20_000, async: 20_000 }, + executionTimeout: { sync: fixture.executionTimeoutMs, async: fixture.executionTimeoutMs }, }), })) vi.mock('@/lib/internal/principals/executor', async (importOriginal) => ({ @@ -3580,6 +3593,337 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( 40_000 ) + it.skipIf(process.env.SIM_HELPERS_SMOKE !== '1' || !process.env.MSHIP_WORKER_ROOT).each([ + // capability, seconds per Wait block (an in-process wait is capped at 300 s). The + // worker waits in 240 s slices: 500 s is a detached run, two run waits and a read-back. + ['declared', 300, 200], + ['undeclared', 260, 0], + // The model's own --wait-timeout 5 lapses: an exit 4 the worker hands the model. + ['own-budget', 30, 0], + ] as const)( + 'draft test run past a wait slice reaches the model settled (capability %s)', + async (capability, waitA, waitB) => { + const workerRoot = process.env.MSHIP_WORKER_ROOT + if (!workerRoot) throw new Error('MSHIP_WORKER_ROOT must point to the worker checkout') + const draftId = generateId() + await db.insert(workflow).values({ + id: draftId, + workspaceId, + userId: 'run-reader', + name: `Long draft ${capability}`, + lastSynced: now, + createdAt: now, + updatedAt: now, + isDeployed: false, + }) + const finishId = generateId() + const blocks: Record = { + start: { + id: 'start', + type: 'start_trigger', + name: 'Start', + position: { x: 0, y: 0 }, + enabled: true, + subBlocks: {}, + outputs: {}, + }, + } + const edges: WorkflowState['edges'] = [] + let previous = 'start' + for (const [index, seconds] of [waitA, waitB].entries()) { + if (seconds <= 0) continue + const id = `wait-${index}` + blocks[id] = { + id, + type: 'wait', + name: `Wait ${index + 1}`, + position: { x: 0, y: 100 * (index + 1) }, + enabled: true, + subBlocks: { + timeValue: { id: 'timeValue', type: 'short-input', value: String(seconds) }, + timeUnit: { id: 'timeUnit', type: 'dropdown', value: 'seconds' }, + }, + outputs: {}, + } + edges.push({ id: `${previous}-${id}`, source: previous, target: id }) + previous = id + } + blocks[finishId] = { + id: finishId, + type: 'function', + name: 'Finish', + position: { x: 0, y: 400 }, + enabled: true, + subBlocks: { + code: { + id: 'code', + type: 'code', + value: 'return { invoice: 42, settled: "after-waits" };', + }, + language: { id: 'language', type: 'dropdown', value: 'javascript' }, + }, + outputs: { result: { type: 'json' } }, + } + edges.push({ id: `${previous}-finish`, source: previous, target: finishId }) + fixture.permission = 'write' + fixture.saved.set(draftId, { blocks, edges, loops: {}, parallels: {} }) + // A paid plan's synchronous limit, so the run outlasts several slices. + fixture.executionTimeoutMs = 50 * 60_000 + + const child = spawn('bun', ['tools/probes/src/draft-run-slices.ts'], { + cwd: workerRoot, + env: { + PATH: process.env.PATH, + NODE_ENV: 'test', + DATABASE_URL: process.env.MSHIP_TEST_DATABASE_URL, + REDIS_URL: process.env.MSHIP_PROBE_REDIS_URL ?? 'redis://127.0.0.1:6379/14', + ANTHROPIC_API_KEY: 'local-scripted-provider', + COPILOT_INBOUND_API_KEY: 'local-controller-probe-key', + SIM_ENDPOINT: controlEndpoint, + INTERNAL_API_SECRET: env.INTERNAL_API_SECRET, + MSHIP_PROBE_WORKFLOW_ID: draftId, + MSHIP_PROBE_RUN_FLAGS: JSON.stringify( + capability === 'own-budget' ? ['--wait-timeout', '5'] : [] + ), + }, + stdio: ['pipe', 'pipe', 'pipe'], + }) + const exited = once(child, 'exit') + const lines = createInterface({ input: child.stdout }) + let diagnostics = '' + child.stderr.on('data', (chunk: Buffer) => { + diagnostics += chunk.toString() + }) + const probeEvents: Array<{ probe: string; at: number }> = [] + const ready = new Promise((resolve) => { + lines.on('line', (line) => { + diagnostics += `${line}\n` + if (!line.startsWith('{"probe":')) return + const message = JSON.parse(line) + if (typeof message.at === 'number') probeEvents.push(message) + if (typeof message.port === 'number') resolve(`http://127.0.0.1:${message.port}`) + }) + }) + const workerRequests: Array<{ path: string; at: number }> = [] + try { + fixture.workerUrl = await Promise.race([ + ready, + exited.then(() => { + throw new Error(`Worker probe exited before readiness: ${diagnostics}`) + }), + ]) + vi.stubEnv('COPILOT_API_KEY', 'local-controller-probe-key') + vi.unstubAllGlobals() + const nativeFetch = globalThis.fetch + vi.stubGlobal('fetch', async (input: RequestInfo | URL, init?: RequestInit) => { + const url = new URL(input instanceof Request ? input.url : input) + if (url.origin !== fixture.workerUrl) + throw new Error(`Unexpected external request: ${url.origin}`) + workerRequests.push({ path: url.pathname, at: Date.now() }) + return nativeFetch(input, init) + }) + ensureHandlersRegistered() + const chatId = generateId() + const messageId = generateId() + await db.insert(copilotChats).values({ + id: chatId, + userId: 'run-reader', + workspaceId, + type: 'mothership', + title: `Long draft ${capability}`, + }) + const startedAt = Date.now() + const result = await runCopilotLifecycle( + { + message: 'Test-run the draft and tell me its output.', + workflowId: draftId, + userId: 'run-reader', + workspaceId, + chatId, + messageId, + // What lib/mothership/chat/payload.ts declares on every chat turn. + ...(capability !== 'undeclared' + ? { clientCapabilities: [WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY] } + : {}), + }, + { + userId: 'run-reader', + workspaceId, + workflowId: draftId, + chatId, + goRoute: '/api/mothership', + interactive: false, + clientToolPickupExpected: false, + flushAfterEvent: false, + abortSignal: AbortSignal.timeout(20 * 60_000), + executionContext: { + userId: 'run-reader', + workspaceId, + workflowId: draftId, + chatId, + userPermission: 'write', + resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry([], { + userId: 'run-reader', + workspaceId, + }), + }, + } + ) + const endedAt = Date.now() + const [run] = await db + .select() + .from(copilotRuns) + .where(eq(copilotRuns.streamId, messageId)) + const calls = run + ? await db + .select() + .from(copilotAsyncToolCalls) + .where(eq(copilotAsyncToolCalls.runId, run.id)) + : [] + const evidence = { + capability, + waits: [waitA, waitB], + startedAt: new Date(startedAt).toISOString(), + totalSeconds: (endedAt - startedAt) / 1000, + success: result.success, + content: result.content, + runStatus: run?.status, + toolCalls: calls + .sort((a, b) => Number(a.createdAt) - Number(b.createdAt)) + .map((call) => ({ + toolCallId: call.toolCallId, + toolName: call.toolName, + args: call.args, + status: call.status, + exitCode: (call.result as { exitCode?: unknown } | null)?.exitCode, + stdout: (call.result as { stdout?: unknown } | null)?.stdout, + startedAt: call.executionStartedAt?.toISOString(), + settledAt: call.executionSettledAt?.toISOString(), + seconds: + call.executionStartedAt && call.executionSettledAt + ? (Number(call.executionSettledAt) - Number(call.executionStartedAt)) / 1000 + : undefined, + })), + workerRequests: workerRequests.map((request) => ({ + ...request, + at: new Date(request.at).toISOString(), + })), + probeEvents, + } + if (process.env.MSHIP_SLICES_REPORT_PATH) + await writeFile( + `${process.env.MSHIP_SLICES_REPORT_PATH}.${capability}.json`, + JSON.stringify(evidence, null, 2) + ) + expect( + result.success, + JSON.stringify({ evidence, errors: fixture.errors, diagnostics }) + ).toBe(true) + expect(run?.status).toBe('complete') + const report = JSON.parse(result.content) + const seen = String(report.output.value) + if (capability === 'own-budget') { + expect(calls).toHaveLength(1) + expect( + (calls[0].args as { request: { invocation: { argv: string[] } } }).request.invocation + .argv + ).toEqual(['workflows', 'run', draftId, '--manual', '--wait-timeout', '5']) + expect(calls[0].status).not.toBe('failed') + expect(seen).toContain('"exitCode":4') + expect(seen).toMatch(/"status": ?"running"/) + expect(seen).toContain('sim workflows runs wait') + await new Promise((resolve) => setTimeout(resolve, 35_000)) + return + } + // Sim #9469: a still-running (exit 4) slice settles as done, not failed. + for (const call of calls) expect(call.status, call.toolName).not.toBe('failed') + // The model saw the settled run, not a detached receipt. + expect(seen).toContain('"exitCode":0') + expect(seen).toContain('"status":"completed"') + expect(seen).toContain('after-waits') + const waitSeconds = waitA + waitB + const argvOf = (call: (typeof calls)[number]) => + (call.args as { request: { invocation: { argv: string[] } } }).request.invocation.argv + const outcome = (call: (typeof calls)[number]) => + call.result as { exitCode: number; stdout: string } + const heldSeconds = (call: (typeof calls)[number]) => + (Number(call.executionSettledAt) - Number(call.executionStartedAt)) / 1000 + const runs = calls.filter((call) => argvOf(call)[1] === 'run') + const waits = calls.filter( + (call) => argvOf(call).slice(0, 3).join(' ') === 'workflows runs wait' + ) + const gets = calls.filter( + (call) => argvOf(call).slice(0, 3).join(' ') === 'workflows runs get' + ) + expect(runs).toHaveLength(1) + expect(calls).toHaveLength(runs.length + waits.length + gets.length) + if (capability === 'declared') { + // First slice: the run itself, bounded at one slice and answered still running. + expect(argvOf(runs[0])).toEqual([ + 'workflows', + 'run', + draftId, + '--manual', + '--wait-timeout', + '240', + ]) + expect(outcome(runs[0]).exitCode).toBe(4) + const receipt = JSON.parse(outcome(runs[0]).stdout) + expect(receipt).toMatchObject({ workflowId: draftId, status: 'running' }) + const runId = receipt.runId + expect(waits).toHaveLength(Math.ceil(waitSeconds / 240) - 1) + for (const [index, call] of waits.entries()) { + expect(argvOf(call)).toEqual([ + 'workflows', + 'runs', + 'wait', + runId, + '--workflow', + draftId, + '--wait-timeout', + '240', + ]) + expect(outcome(call).exitCode).toBe(index === waits.length - 1 ? 0 : 4) + } + expect(JSON.parse(outcome(waits.at(-1)!).stdout)).toMatchObject({ + runId, + status: 'completed', + }) + // Settled: read back once with its output, as the held call printed it. + expect(gets).toHaveLength(1) + expect(argvOf(gets[0])).toEqual([ + 'workflows', + 'runs', + 'get', + runId, + '--workflow', + draftId, + '--include-output', + ]) + expect(outcome(gets[0]).stdout).toContain('after-waits') + // No single Sim call held the whole run. + for (const call of calls) expect(heldSeconds(call)).toBeLessThan(250) + } else { + expect(argvOf(runs[0])).toEqual(['workflows', 'run', draftId, '--manual']) + expect(waits).toHaveLength(0) + expect(gets).toHaveLength(0) + expect(outcome(runs[0]).exitCode).toBe(0) + // One held call for the whole run, as on staging. + expect(heldSeconds(runs[0])).toBeGreaterThan(waitSeconds) + } + expect(endedAt - startedAt).toBeGreaterThan(waitSeconds * 1000) + } finally { + fixture.executionTimeoutMs = 20_000 + child.kill() + await exited + lines.close() + vi.unstubAllGlobals() + vi.unstubAllEnvs() + } + }, + 25 * 60_000 + ) + it .skipIf(process.env.SIM_HELPERS_SMOKE !== '1' || !process.env.MSHIP_WORKER_ROOT) .each(['valid', 'rewrite'] as const)( From c0c2b6c3855fa517ca1c8562e9b50e36c14ca252 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 30 Oct 2026 10:14:26 -0700 Subject: [PATCH 2/2] test(mothership): keep the draft-run case inside the repo's audits Uses sleep from @sim/utils/helpers, guards the last wait slice instead of a non-null assertion, gives the probe's startup its own ten-second deadline like the neighbouring controller case, and tells knip the probe path lives in the worker checkout. --- .../mothership/agent-cli/saved-run-read.live.test.ts | 10 ++++++++-- knip.jsonc | 7 +++++-- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts b/apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts index 9ab6bdfc9ad..a835b935e38 100644 --- a/apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts +++ b/apps/sim/lib/mothership/agent-cli/saved-run-read.live.test.ts @@ -63,6 +63,7 @@ import { } from '@sim/db/schema' import type { PermissionType } from '@sim/platform-authz/workspace' import { authMockFns, envFlagsMock, setEnvFlags } from '@sim/testing' +import { sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import { eq, is, SQL, sql } from 'drizzle-orm' import { getTableConfig } from 'drizzle-orm/pg-core' @@ -3705,6 +3706,7 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( }) }) const workerRequests: Array<{ path: string; at: number }> = [] + const startupTimeout = setTimeout(() => child.kill(), 10_000) try { fixture.workerUrl = await Promise.race([ ready, @@ -3712,6 +3714,7 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( throw new Error(`Worker probe exited before readiness: ${diagnostics}`) }), ]) + clearTimeout(startupTimeout) vi.stubEnv('COPILOT_API_KEY', 'local-controller-probe-key') vi.unstubAllGlobals() const nativeFetch = globalThis.fetch @@ -3832,7 +3835,7 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( expect(seen).toContain('"exitCode":4') expect(seen).toMatch(/"status": ?"running"/) expect(seen).toContain('sim workflows runs wait') - await new Promise((resolve) => setTimeout(resolve, 35_000)) + await sleep(35_000) return } // Sim #9469: a still-running (exit 4) slice settles as done, not failed. @@ -3885,7 +3888,9 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( ]) expect(outcome(call).exitCode).toBe(index === waits.length - 1 ? 0 : 4) } - expect(JSON.parse(outcome(waits.at(-1)!).stdout)).toMatchObject({ + const settledWait = waits.at(-1) + if (!settledWait) throw new Error('The run never settled through a wait slice') + expect(JSON.parse(outcome(settledWait).stdout)).toMatchObject({ runId, status: 'completed', }) @@ -3913,6 +3918,7 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( } expect(endedAt - startedAt).toBeGreaterThan(waitSeconds * 1000) } finally { + clearTimeout(startupTimeout) fixture.executionTimeoutMs = 20_000 child.kill() await exited diff --git a/knip.jsonc b/knip.jsonc index 852165006cf..562cef620f8 100644 --- a/knip.jsonc +++ b/knip.jsonc @@ -93,8 +93,11 @@ // Knip's email CLI plugin infers this transitive preview implementation. "@react-email/ui" ], - // Live test spawns this path with cwd=MSHIP_WORKER_ROOT (a sibling repo). - "ignoreUnresolved": ["tools/probes/src/controller-run-read.ts"] + // Live tests spawn these paths with cwd=MSHIP_WORKER_ROOT (a sibling repo). + "ignoreUnresolved": [ + "tools/probes/src/controller-run-read.ts", + "tools/probes/src/draft-run-slices.ts" + ] }, "apps/realtime": { "includeEntryExports": false,