From 424459edd795478b77bec0efffb69d1123c0d42a Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Fri, 30 Oct 2026 23:37:43 -0700 Subject: [PATCH 01/14] feat(mothership): answer a workflow tool's run still running past the worker's wait A run_workflow* call held the turn until its run settled, up to the execution limit, while a synchronous workflows run can already detach after --wait-timeout (#9464). Sim now declares workflow-tool-wait-timeout and honors the waitTimeoutMs a workflow tool frame names (synced from the mothership contract): - Server-run call: the handler gives the run its execution id up front and, once the wait passes, answers {status: running, executionId, workflowId}. The run goes on in-process; the tool execution keeps holding it, so its lease and a Stop through the call's abort signal still reach it. - Browser-run call: once the wait passes and a browser holds the call's execution claim, the call is settled background, the detach a browser sends on pagehide. The browser's run goes on, still bound to the call, so a Stop cancels it. - A background workflow completion now carries status: running beside its ids, the shape the worker waits on. A frame without waitTimeoutMs, or one with a malformed value, holds the call as before. --- .../sim/app/api/copilot/confirm/route.test.ts | 2 +- apps/sim/lib/mothership/chat/payload.test.ts | 4 + apps/sim/lib/mothership/chat/payload.ts | 9 +- apps/sim/lib/mothership/generated/protocol.ts | 31 ++- .../request/handlers/handlers.test.ts | 29 +++ .../lib/mothership/request/handlers/tool.ts | 6 + .../mothership/request/tools/client.test.ts | 3 +- .../lib/mothership/request/tools/deadline.ts | 14 ++ .../lib/mothership/request/tools/executor.ts | 6 +- .../tools/workflow-client-fallback.test.ts | 92 +++++++- .../request/tools/workflow-client-fallback.ts | 75 +++++- apps/sim/lib/mothership/request/types.ts | 5 + .../sim/lib/mothership/tool-executor/types.ts | 5 + .../tools/handlers/workflow/mutations.test.ts | 112 ++++++++- .../tools/handlers/workflow/mutations.ts | 223 ++++++++++++------ .../lib/mothership/tools/workflow-tools.ts | 3 + .../application/run-workflow-from-copilot.ts | 8 +- scripts/check-unused-exports.baseline.json | 1 - 18 files changed, 544 insertions(+), 84 deletions(-) diff --git a/apps/sim/app/api/copilot/confirm/route.test.ts b/apps/sim/app/api/copilot/confirm/route.test.ts index a8a41d6bf64..6f7fef6baf0 100644 --- a/apps/sim/app/api/copilot/confirm/route.test.ts +++ b/apps/sim/app/api/copilot/confirm/route.test.ts @@ -1036,7 +1036,7 @@ describe('Copilot Confirm API Route', () => { status: 'background', message: 'Workflow execution is continuing in the background.', timestamp: expect.any(String), - data: { workflowId: 'workflow-1', executionId: 'unbound-execution' }, + data: { workflowId: 'workflow-1', executionId: 'unbound-execution', status: 'running' }, }) }) diff --git a/apps/sim/lib/mothership/chat/payload.test.ts b/apps/sim/lib/mothership/chat/payload.test.ts index c82da7f3b93..b300f46a5d3 100644 --- a/apps/sim/lib/mothership/chat/payload.test.ts +++ b/apps/sim/lib/mothership/chat/payload.test.ts @@ -34,6 +34,7 @@ import { ChatPayloadSchema, TOOL_DEADLINE_CAPABILITY, WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY, } from '@/lib/mothership/generated/protocol' import { searchUsersV2Tool } from '@/tools/github/search_users' import { gmailListLabelsV2Tool } from '@/tools/gmail/list_labels' @@ -430,6 +431,7 @@ describe('buildCopilotRequestPayload', () => { 'workflow-tool-pickup', TOOL_DEADLINE_CAPABILITY, WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY, ]) } ) @@ -514,6 +516,7 @@ describe('buildCopilotRequestPayload', () => { expect(payload.clientCapabilities).toEqual([ TOOL_DEADLINE_CAPABILITY, WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY, ]) if (owner === 'organization') expect(payload).not.toHaveProperty('integrationTools') } @@ -854,6 +857,7 @@ describe('Assistant payload', () => { expect(payload.clientCapabilities).toEqual([ TOOL_DEADLINE_CAPABILITY, WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY, ]) expect(payload.mode).toBe('assistant') expect(payload.assistantSearch).toEqual({ source: 'slack', documentIds: ['document-1'] }) diff --git a/apps/sim/lib/mothership/chat/payload.ts b/apps/sim/lib/mothership/chat/payload.ts index 33b758e3afb..a6cace00771 100644 --- a/apps/sim/lib/mothership/chat/payload.ts +++ b/apps/sim/lib/mothership/chat/payload.ts @@ -38,6 +38,7 @@ import { type ModelSelection, TOOL_DEADLINE_CAPABILITY, WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY, } from '@/lib/mothership/generated/protocol' import type { VfsSnapshotV1 } from '@/lib/mothership/generated/vfs-snapshot-v1' import { getToolEntry } from '@/lib/mothership/tool-executor/router' @@ -562,13 +563,15 @@ export async function buildCopilotRequestPayload(params: BuildPayloadParams): Pr : {}), // The mounted chat view executes client-routed workflow tools (run panel UX), so the // UI declares that capability explicitly; headless callers omit it and the server runs - // those tools immediately. The deadline and run-wait tokens are this server's own: it - // bounds every handed call by the deadline its frame names, and its CLI lets a - // synchronous `workflows run` outlast `--wait-timeout`. + // those tools immediately. The deadline and wait tokens are this server's own: it + // bounds every handed call by the deadline its frame names, its CLI lets a synchronous + // `workflows run` outlast `--wait-timeout`, and a workflow tool's run outlasts the + // `waitTimeoutMs` its frame names. clientCapabilities: [ ...(isAssistant ? [] : ['workflow-tool-pickup']), TOOL_DEADLINE_CAPABILITY, WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY, ], } } diff --git a/apps/sim/lib/mothership/generated/protocol.ts b/apps/sim/lib/mothership/generated/protocol.ts index 2b252886dde..64b473165f1 100644 --- a/apps/sim/lib/mothership/generated/protocol.ts +++ b/apps/sim/lib/mothership/generated/protocol.ts @@ -297,6 +297,14 @@ export interface StreamToolReplay { */ export interface StreamToolDeadline { deadlineMs?: number | undefined; + /** + * For a workflow tool (`run_workflow`, `run_workflow_until_block`, `run_from_block`, + * `run_block`) on a Sim that declares {@link WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY}: how long + * Sim waits for the call's run to settle once it starts waiting on it. A run still going then + * keeps going and the call settles with a {@link WorkflowToolRunningResult}. Absent = held + * until the run settles. + */ + waitTimeoutMs?: number | undefined; } /** @@ -322,6 +330,24 @@ export const WORKSPACE_FILE_COPY_CAPABILITY = "workspace-file-copy"; */ export const WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY = "workflow-run-wait-timeout"; +/** + * The `clientCapabilities` token a Sim sends when it honors `StreamToolDeadline.waitTimeoutMs` + * on a workflow tool call: a run that outlasts it continues and the call settles with a + * {@link WorkflowToolRunningResult}, so the worker waits on the run in slices. + */ +export const WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY = "workflow-tool-wait-timeout"; + +/** + * A workflow tool call's output when its run outlasted the call's `waitTimeoutMs`: the run is + * still going under `executionId`, readable as a run of `workflowId` (`workflows runs wait`). + */ +export interface WorkflowToolRunningResult { + status: "running"; + executionId: string; + workflowId: string; + message?: string | undefined; +} + /** POST /api/mothership — the chat request sim sends. */ export interface ChatRequest extends StreamResponseReceipt { fileOwnerProtocolVersion?: 1 | undefined; @@ -367,8 +393,9 @@ export interface ChatRequest extends StreamResponseReceipt { * grace windows and run tools server-side immediately. ABSENT = older/unknown caller — * dispatch keeps its conservative grace (deploy-skew safe: a stale tab that predates * this field still gets waited on). Known capability: "workflow-tool-pickup". Sim also - * declares its own {@link TOOL_DEADLINE_CAPABILITY}, {@link WORKSPACE_FILE_COPY_CAPABILITY} and - * {@link WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY} here. + * declares its own {@link TOOL_DEADLINE_CAPABILITY}, {@link WORKSPACE_FILE_COPY_CAPABILITY}, + * {@link WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY} and {@link WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY} + * here. */ clientCapabilities?: string[] | undefined; /** diff --git a/apps/sim/lib/mothership/request/handlers/handlers.test.ts b/apps/sim/lib/mothership/request/handlers/handlers.test.ts index 89127e9d8e0..3d1ab77e7d2 100644 --- a/apps/sim/lib/mothership/request/handlers/handlers.test.ts +++ b/apps/sim/lib/mothership/request/handlers/handlers.test.ts @@ -732,6 +732,35 @@ describe('sse-handlers tool lifecycle', () => { expect(context.toolCalls.get('takeover-1')?.deadlineAt).toBeLessThan(before + 60_000) }) + it('hands a workflow tool the wait its frame names, and no other tool', async () => { + executeTool.mockResolvedValue({ success: true, output: {} }) + const frame = (toolCallId: string, toolName: string, waitTimeoutMs: unknown) => { + const event = deadlineFrame(900_000, { + toolCallId, + toolName, + arguments: { workflowId: 'workflow-1' }, + }) + return { ...event, payload: { ...event.payload, waitTimeoutMs } } as StreamEvent + } + await sseHandlers.tool(frame('run-1', 'run_workflow', 240_000), context, execContext, { + interactive: false, + }) + await sseHandlers.tool(frame('audio-2', 'generate_audio', 240_000), context, execContext, { + interactive: false, + }) + await sseHandlers.tool(frame('run-2', 'run_workflow', -5), context, execContext, { + interactive: false, + }) + await Promise.all(context.pendingToolPromises.values()) + const handed = (toolCallId: string) => + executeTool.mock.calls.find( + ([, , toolContext]) => toolContext?.toolCallId === toolCallId + )?.[2] + expect(handed('run-1')?.workflowRunWaitMs).toBe(240_000) + expect(handed('audio-2')).not.toHaveProperty('workflowRunWaitMs') + expect(handed('run-2')).not.toHaveProperty('workflowRunWaitMs') + }) + /** Runs a desktop-local read whose frame names `deadlineMs`, recording the emitted events. */ async function runLocalRead(deadlineMs: number) { const onEvent = vi.fn() diff --git a/apps/sim/lib/mothership/request/handlers/tool.ts b/apps/sim/lib/mothership/request/handlers/tool.ts index 6955e09ba01..3ec10080dae 100644 --- a/apps/sim/lib/mothership/request/handlers/tool.ts +++ b/apps/sim/lib/mothership/request/handlers/tool.ts @@ -44,6 +44,7 @@ import { sealClientToolContext } from '@/lib/mothership/request/tools/client-com import { capToToolDeadline, frameToolDeadlineAt, + frameWorkflowWaitMs, TOOL_DEADLINE_BEFORE_START_MESSAGE, TOOL_DEADLINE_EXCEEDED_MESSAGE, toDeadlineDate, @@ -742,6 +743,8 @@ async function handleCallPhase( if (parentToolCallId) toolCall.parentToolCallId = parentToolCallId if (data.execName) toolCall.execName = data.execName toolCall.deadlineAt ??= frameToolDeadlineAt(data.deadlineMs, Date.now()) + if (isWorkflowToolName(toolName)) + toolCall.waitTimeoutMs ??= frameWorkflowWaitMs(data.waitTimeoutMs) const readPath = typeof args?.path === 'string' ? args.path : undefined if (toolName === 'read' && readPath?.startsWith('internal/')) return @@ -1110,6 +1113,9 @@ async function dispatchToolExecution( options.clientToolPickupExpected === false ? 0 : COPILOT_WORKFLOW_TOOL_CLIENT_GRACE_MS, + ...(toolCall.waitTimeoutMs !== undefined + ? { waitTimeoutMs: toolCall.waitTimeoutMs } + : {}), abortSignal: options.abortSignal, registry: execContext.resolvedSecretTraceRegistry, runOnServer: (boundExecutionId) => { diff --git a/apps/sim/lib/mothership/request/tools/client.test.ts b/apps/sim/lib/mothership/request/tools/client.test.ts index 7ba230faee4..ff20d09a7b9 100644 --- a/apps/sim/lib/mothership/request/tools/client.test.ts +++ b/apps/sim/lib/mothership/request/tools/client.test.ts @@ -729,7 +729,8 @@ describe('workflow client tool completion', () => { expect(completion).toEqual({ status: 'background', message: 'Workflow execution is continuing in the background.', - data: { workflowId: 'workflow-1', executionId: 'execution-1' }, + // A WorkflowToolRunningResult: the worker waits on the run by these ids. + data: { workflowId: 'workflow-1', executionId: 'execution-1', status: 'running' }, }) expect(registry.isComplete()).toBe(true) expect(getTrustedWorkflowToolExecution).not.toHaveBeenCalled() diff --git a/apps/sim/lib/mothership/request/tools/deadline.ts b/apps/sim/lib/mothership/request/tools/deadline.ts index 4a013afdcc8..3df01bea69a 100644 --- a/apps/sim/lib/mothership/request/tools/deadline.ts +++ b/apps/sim/lib/mothership/request/tools/deadline.ts @@ -20,6 +20,20 @@ export function frameToolDeadlineAt(deadlineMs: unknown, receivedAt: number): nu return receivedAt + deadlineMs - TOOL_DEADLINE_SETTLE_MARGIN_MS } +/** + * How long a workflow tool call waits for its run before settling still running, from the + * frame's `waitTimeoutMs`. Undefined when the frame names none or a malformed value: the call + * holds until its run settles. + */ +export function frameWorkflowWaitMs(waitTimeoutMs: unknown): number | undefined { + return typeof waitTimeoutMs === 'number' && + Number.isSafeInteger(waitTimeoutMs) && + waitTimeoutMs > 0 && + waitTimeoutMs <= MAX_FRAME_DEADLINE_MS + ? waitTimeoutMs + : undefined +} + /** What the model learns when a call's deadline passed before the call could start. */ export const TOOL_DEADLINE_BEFORE_START_MESSAGE = 'This tool call ran out of its time limit before it could start, so it did not run. Nothing changed; it is safe to try again with a smaller request.' diff --git a/apps/sim/lib/mothership/request/tools/executor.ts b/apps/sim/lib/mothership/request/tools/executor.ts index c3fba27e047..4ea8030b5ee 100644 --- a/apps/sim/lib/mothership/request/tools/executor.ts +++ b/apps/sim/lib/mothership/request/tools/executor.ts @@ -369,13 +369,17 @@ class ToolExecutionTimeoutError extends Error { /** Builds the per-call context from the turn-scoped execution context. */ export function buildToolExecutionContext( - toolCall: Pick, + toolCall: Pick< + ToolCallState, + 'id' | 'parentToolCallId' | 'params' | 'targetWorkspaceId' | 'waitTimeoutMs' + >, execContext: ExecutionContext ): ExecutionContext { return { ...execContext, toolCallId: toolCall.id, ...(toolCall.targetWorkspaceId ? { targetWorkspaceId: toolCall.targetWorkspaceId } : {}), + ...(toolCall.waitTimeoutMs !== undefined ? { workflowRunWaitMs: toolCall.waitTimeoutMs } : {}), resolvedSecretTraceRegistry: execContext.resolvedSecretTraceRegistry?.forkForInputPaths([]), ...(toolCall.parentToolCallId ? { parentToolCallId: toolCall.parentToolCallId } : {}), } diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.test.ts b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.test.ts index ec29de68d56..13e50f4c49d 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.test.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.test.ts @@ -8,8 +8,13 @@ import { } from '@sim/testing/mocks/mothership-client-tool-waiter.mock' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { recordDegraded } = vi.hoisted(() => ({ +const { recordDegraded, settleClientToolCall } = vi.hoisted(() => ({ recordDegraded: vi.fn(), + settleClientToolCall: vi.fn(), +})) + +vi.mock('@/lib/mothership/request/tools/client-settlement.server', () => ({ + settleClientToolCall, })) vi.mock('@/lib/mothership/request/metrics', () => ({ recordDegraded })) @@ -24,6 +29,8 @@ vi.mock('@/lib/mothership/async-runs/repository', () => mothershipAsyncRunsMock) import { raceWorkflowToolClientPickup } from '@/lib/mothership/request/tools/workflow-client-fallback' const claimWorkflowToolExecution = mothershipAsyncRunsMockFns.mockClaimWorkflowToolExecution +const getAsyncToolCall = mothershipAsyncRunsMockFns.mockGetAsyncToolCall +const getClaimedWorkflowExecutionId = mothershipAsyncRunsMockFns.mockGetClaimedWorkflowExecutionId const GRACE_MS = 30_000 const TIMEOUT_MS = 3_600_000 @@ -216,3 +223,86 @@ describe('raceWorkflowToolClientPickup', () => { expect(params.runOnServer).not.toHaveBeenCalled() }) }) + +describe('a browser-run workflow call past its wait', () => { + const WAIT_MS = 240_000 + beforeEach(() => { + vi.useFakeTimers() + waiterSignals = [] + settleClientToolCall.mockReset() + getAsyncToolCall.mockReset() + getClaimedWorkflowExecutionId.mockImplementation((claimedBy?: string | null) => + claimedBy?.startsWith('workflow:') ? claimedBy.slice('workflow:'.length) : undefined + ) + pendingUntilAborted() + // A browser already holds the call, so the server keeps waiting on it. + claimWorkflowToolExecution.mockResolvedValue(null) + }) + afterEach(() => { + vi.useRealTimers() + }) + + const browserClaim = { + status: 'running', + claimedBy: 'workflow:execution-1', + clientWorkflowExecutionId: 'execution-1', + } + + it('settles the call running in the background by the browser run ids', async () => { + getAsyncToolCall.mockResolvedValue(browserClaim) + const params = baseParams({ waitTimeoutMs: WAIT_MS }) + void raceWorkflowToolClientPickup(params as never) + + await vi.advanceTimersByTimeAsync(WAIT_MS - 1) + expect(settleClientToolCall).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(1) + + expect(settleClientToolCall).toHaveBeenCalledWith({ + toolCallId: 'tool-1', + status: 'background', + message: 'Workflow execution is continuing in the background.', + data: { workflowId: 'workflow-1', executionId: 'execution-1', status: 'running' }, + executionId: 'execution-1', + guard: { kind: 'open' }, + }) + // The browser's run goes on; only the server's wait ended. + expect(params.runOnServer).not.toHaveBeenCalled() + }) + + it('re-reads a call no browser has claimed yet until one does', async () => { + getAsyncToolCall + .mockResolvedValueOnce({ + status: 'running', + claimedBy: null, + clientWorkflowExecutionId: null, + }) + .mockResolvedValue(browserClaim) + void raceWorkflowToolClientPickup(baseParams({ waitTimeoutMs: WAIT_MS }) as never) + + await vi.advanceTimersByTimeAsync(WAIT_MS) + expect(settleClientToolCall).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(5_000) + expect(settleClientToolCall).toHaveBeenCalledWith( + expect.objectContaining({ status: 'background', executionId: 'execution-1' }) + ) + }) + + it('leaves a call that settled first alone', async () => { + waitForWorkflowToolCompletion.mockResolvedValue({ status: 'success', data: { ok: true } }) + const outcome = await raceWorkflowToolClientPickup( + baseParams({ waitTimeoutMs: WAIT_MS }) as never + ) + await vi.advanceTimersByTimeAsync(WAIT_MS * 2) + expect(outcome.winner).toBe('client') + expect(getAsyncToolCall).not.toHaveBeenCalled() + expect(settleClientToolCall).not.toHaveBeenCalled() + }) + + it('holds a call that names no wait', async () => { + getAsyncToolCall.mockResolvedValue(browserClaim) + void raceWorkflowToolClientPickup(baseParams() as never) + await vi.advanceTimersByTimeAsync(TIMEOUT_MS / 2) + expect(getAsyncToolCall).not.toHaveBeenCalled() + expect(settleClientToolCall).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts index 0c346b84418..0d2dac3eb47 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts @@ -6,11 +6,24 @@ import type { AsyncCompletionSignal, AsyncTerminalCompletionSnapshot, } from '@/lib/mothership/async-runs/lifecycle' -import { claimWorkflowToolExecution } from '@/lib/mothership/async-runs/repository' +import { + ASYNC_TOOL_CONFIRMATION_STATUS, + ASYNC_TOOL_STATUS, +} from '@/lib/mothership/async-runs/lifecycle' +import { + claimWorkflowToolExecution, + getAsyncToolCall, + getClaimedWorkflowExecutionId, +} from '@/lib/mothership/async-runs/repository' import { CopilotDegradedReason } from '@/lib/mothership/generated/trace-attribute-values-v1' import { recordDegraded } from '@/lib/mothership/request/metrics' import { waitForWorkflowToolCompletion } from '@/lib/mothership/request/tools/client' +import { settleClientToolCall } from '@/lib/mothership/request/tools/client-settlement.server' import { toolStartWindowClosed } from '@/lib/mothership/request/tools/deadline' +import { + createStructuralWorkflowToolCompletionData, + getWorkflowToolCompletionMessage, +} from '@/lib/mothership/tools/workflow-tools' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' const logger = createLogger('CopilotWorkflowClientFallback') @@ -36,12 +49,56 @@ interface RaceWorkflowToolClientPickupParams { /** When the call's deadline passes: past it the server does not take the call over. */ deadlineAt?: number graceMs: number + /** + * How long to wait for a browser's run before settling the call still running (the frame's + * `waitTimeoutMs`); the browser's run goes on. Undefined waits until it settles. + */ + waitTimeoutMs?: number abortSignal?: AbortSignal registry?: ResolvedSecretTraceRegistry /** Runs the tool in-process; only invoked after the execution claim is won. */ runOnServer: (boundExecutionId: string) => Promise } +/** How often a lapsed wait re-reads the call for the browser's claim. */ +const CLIENT_CLAIM_POLL_MS = 5000 + +/** + * Once `waitTimeoutMs` has passed, settles a browser-run workflow call as running in the + * background, the same detach a browser sends on `pagehide`: its waiter answers the call with + * the run's ids and the browser's run goes on, bound to the call, so a Stop still cancels it. + * A call no browser has claimed yet is re-read until one does or the wait ends. + */ +async function detachClientRunAfterWait(params: { + toolCallId: string + workflowId?: string + waitTimeoutMs: number + signal: AbortSignal +}): Promise { + const { toolCallId, workflowId, signal } = params + if (!workflowId) return + let delayMs = params.waitTimeoutMs + while (!signal.aborted) { + await sleep(delayMs) + delayMs = CLIENT_CLAIM_POLL_MS + if (signal.aborted) return + const row = await getAsyncToolCall(toolCallId) + if (!row || row.status !== ASYNC_TOOL_STATUS.running) return + const executionId = getClaimedWorkflowExecutionId(row.claimedBy) + if (!executionId || row.clientWorkflowExecutionId !== executionId) continue + const status = ASYNC_TOOL_CONFIRMATION_STATUS.background + await settleClientToolCall({ + toolCallId, + status, + message: getWorkflowToolCompletionMessage(status), + data: createStructuralWorkflowToolCompletionData(status, workflowId, executionId), + executionId, + guard: { kind: 'open' }, + }) + return + } +} + /** * Wait for a browser to run a workflow tool call, and run it here if none does. * @@ -81,6 +138,22 @@ export async function raceWorkflowToolClientPickup( abortSignal: clientWaitSignal, registry, }) + if (params.waitTimeoutMs !== undefined) { + const stopDetach = new AbortController() + void clientWait.finally(() => stopDetach.abort()) + detachClientRunAfterWait({ + toolCallId, + workflowId, + waitTimeoutMs: params.waitTimeoutMs, + signal: AbortSignal.any([clientWaitSignal, stopDetach.signal]), + }).catch((error) => { + logger.warn('Could not settle a lapsed browser workflow wait as running', { + toolCallId, + workflowId, + error: toError(error).message, + }) + }) + } // A caller-supplied timeout shorter than the grace must win, or the grace // would outlive the wait it is supposed to bound. diff --git a/apps/sim/lib/mothership/request/types.ts b/apps/sim/lib/mothership/request/types.ts index d96ed2ea5b8..6792e5b560e 100644 --- a/apps/sim/lib/mothership/request/types.ts +++ b/apps/sim/lib/mothership/request/types.ts @@ -63,6 +63,11 @@ export interface ToolCallState { * the worker named none. */ deadlineAt?: number + /** + * For a workflow tool, how long to wait for its run to settle before settling the call still + * running (the frame's `waitTimeoutMs`). Undefined holds the call until the run settles. + */ + waitTimeoutMs?: number /** * For a subagent-scoped tool call, the invoking subagent's channel id (its * outer tool_use id, = event.scope.parentToolCallId). Captured at dispatch so diff --git a/apps/sim/lib/mothership/tool-executor/types.ts b/apps/sim/lib/mothership/tool-executor/types.ts index d0f5c40d4a1..ab6e610b131 100644 --- a/apps/sim/lib/mothership/tool-executor/types.ts +++ b/apps/sim/lib/mothership/tool-executor/types.ts @@ -35,6 +35,11 @@ export interface ToolExecutionContext { * into the principal by `requireTrustedCopilotExecutionContext`. */ boundWorkflowExecutionId?: string + /** + * How long a workflow tool waits for its run before answering it still running and letting it + * continue (the call frame's `waitTimeoutMs`). Undefined waits until the run settles. + */ + workflowRunWaitMs?: number billingAttribution?: BillingAttributionSnapshot copilotToolExecution?: boolean /** In-process authoring admission; never supplied by model arguments or checkpoint JSON. */ 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 658e3900d07..9c5647a78ca 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts @@ -1,7 +1,8 @@ import { executeWorkflowMock } from '@sim/testing/mocks/execute-workflow.mock' import { telemetryMock } from '@sim/testing/mocks/telemetry.mock' import { workflowsOrchestrationMock } from '@sim/testing/mocks/workflows-orchestration.mock' -import { beforeEach, describe, expect, it, vi } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { observeSandboxExecution } from '@/lib/execution/remote-sandbox/observer' import { WorkflowRunAlreadyTerminalError } from '@/lib/execution/workflow-run-already-terminal-error' import type { ExecutionContext } from '@/lib/mothership/request/types' import { executeTool } from '@/lib/mothership/tool-executor/executor' @@ -427,3 +428,112 @@ describe('workflow mutation Copilot adapters', () => { expect((await run()).effect).toEqual(effect) }) }) + +describe('a workflow tool run past its wait', () => { + beforeEach(() => { + vi.useFakeTimers() + mocks.hasExecutionResult.mockReturnValue(false) + mocks.readAttemptedExecutionId.mockReturnValue(undefined) + }) + afterEach(() => { + vi.useRealTimers() + }) + + /** A run the test settles itself, and the lifecycle it was started with. */ + function pendingRun() { + let settle: (value: unknown) => void = () => {} + const started = new Promise((resolve) => { + settle = resolve + }) + mocks.executeWorkflowUseCase.mockReturnValueOnce(started) + return { + settle, + lifecycle: () => + mocks.executeWorkflowUseCase.mock.calls.at(-1)?.[2].lifecycle as { + executionId?: string + }, + } + } + + it('answers still running by the id it gave the run, and holds the run until it settles', async () => { + const run = pendingRun() + const held: Promise[] = [] + const answer = observeSandboxExecution( + { + hold: (work) => { + held.push(work) + }, + unsettled: () => {}, + }, + () => + executeRunWorkflow({ workflowId: 'workflow-1' }, { ...context, workflowRunWaitMs: 240_000 }) + ) + await vi.advanceTimersByTimeAsync(240_000) + const result = await answer + const executionId = run.lifecycle().executionId + expect(executionId).toEqual(expect.any(String)) + expect(result).toEqual({ + success: true, + output: { + status: 'running', + executionId, + workflowId: 'workflow-1', + message: expect.stringContaining(String(executionId)), + }, + effect: { phase: 'attempted', ids: { executionId } }, + }) + // The tool's execution still owns the run, so its lease and Stop reach it. + expect(held).toHaveLength(1) + run.settle({ success: true, output: { done: true }, logs: [], metadata: { executionId } }) + await expect(held[0]).resolves.toMatchObject({ success: true }) + }) + + it('answers with the result when the run settles within its wait', async () => { + const run = pendingRun() + const answer = executeRunBlock( + { workflowId: 'workflow-1', blockId: 'block-1' }, + { ...context, workflowRunWaitMs: 240_000 } + ) + await vi.advanceTimersByTimeAsync(1_000) + run.settle({ + success: true, + output: { value: 7 }, + logs: [], + metadata: { executionId: run.lifecycle().executionId }, + }) + await expect(answer).resolves.toMatchObject({ + success: true, + output: { output: { value: 7 } }, + }) + }) + + it('keeps the bound execution id of a server-run call', async () => { + const run = pendingRun() + const answer = executeRunWorkflow( + { workflowId: 'workflow-1' }, + { ...context, boundWorkflowExecutionId: 'bound-1', workflowRunWaitMs: 1_000 } + ) + await vi.advanceTimersByTimeAsync(1_000) + await expect(answer).resolves.toMatchObject({ + output: { status: 'running', executionId: 'bound-1' }, + }) + expect(run.lifecycle()).toMatchObject({ + boundExecution: { executionId: 'bound-1', copilotToolCallId: 'tool-call-1' }, + }) + run.settle({ success: true, output: {}, logs: [] }) + }) + + it('holds the call until the run settles when the call names no wait', async () => { + const run = pendingRun() + let answered = false + const answer = executeRunWorkflow({ workflowId: 'workflow-1' }, context).then((result) => { + answered = true + return result + }) + await vi.advanceTimersByTimeAsync(60 * 60_000) + expect(answered).toBe(false) + expect(run.lifecycle().executionId).toBeUndefined() + run.settle({ success: true, output: { value: 1 }, logs: [], metadata: { executionId: 'e-1' } }) + await expect(answer).resolves.toMatchObject({ success: true, output: { output: { value: 1 } } }) + }) +}) diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts index 81cf7ce552f..08da7649253 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts @@ -1,13 +1,16 @@ import { createLogger } from '@sim/logger' +import { generateId } from '@sim/utils/id' import { filterUndefined, isPlainRecord, isRecordLike } from '@sim/utils/object' import { createCopilotWorkspaceApiKey } from '@/lib/api-key/application/create-api-key' import { PlatformEvents } from '@/lib/core/telemetry' +import { retainSandboxExecution } from '@/lib/execution/remote-sandbox/observer' import { messageForCopilotApplicationError } from '@/lib/mothership/application/errors' import { executeCopilotApiKeyUseCase } from '@/lib/mothership/application/execute-api-key-use-case' import { executeCopilotWorkflowUseCase, messageForCopilotWorkflowError, } from '@/lib/mothership/application/execute-workflow-use-case' +import type { WorkflowToolRunningResult } from '@/lib/mothership/generated/protocol' import type { ExecutionContext, ToolCallResult } from '@/lib/mothership/request/types' import { TOOL_EFFECT_PHASE, @@ -29,6 +32,7 @@ import { } from '@/lib/mothership/tools/workflow-output' import { cancelWorkflowRun } from '@/lib/workflows/application/cancel-run' import { + type CopilotWorkflowRunLifecycle, runBlockFromCopilot, runFromBlockFromCopilot, runWorkflowFromCopilot, @@ -252,6 +256,52 @@ function copilotRunLifecycle(context: ExecutionContext) { } } +/** + * Runs a workflow tool's run and answers with its result. When the call names a wait the run + * outlasts (`workflowRunWaitMs`), it answers the run still running by the execution id it gave + * the run, and the run goes on. The tool's execution keeps holding the run, so its lease, and a + * Stop through the call's abort signal, reach the run until it settles. + */ +async function answerWithinWait( + context: ExecutionContext, + workflowId: string, + run: (lifecycle: CopilotWorkflowRunLifecycle) => Promise +): Promise { + const lifecycle = copilotRunLifecycle(context) + const waitMs = context.workflowRunWaitMs + if (waitMs === undefined) return run(lifecycle) + const executionId = lifecycle.boundExecution?.executionId ?? generateId() + const settled = run({ ...lifecycle, executionId }) + retainSandboxExecution(settled) + let timer: ReturnType | undefined + const waited = new Promise((resolve) => { + timer = setTimeout(() => resolve(null), waitMs) + }) + try { + const result = await Promise.race([settled, waited]) + if (result) return result + } finally { + clearTimeout(timer) + } + logger.info('Workflow tool run outlasted its wait; answering it still running', { + toolCallId: context.toolCallId, + workflowId, + executionId, + waitMs, + }) + const output: WorkflowToolRunningResult = { + status: 'running', + executionId, + workflowId, + message: `The run is still going after ${Math.round(waitMs / 1000)} seconds and continues in the background as execution ${executionId}.`, + } + return { + success: true, + output, + effect: executionEffect(TOOL_EFFECT_PHASE.attempted, executionId), + } +} + function assertWorkflowMutationNotAborted( context: ExecutionContext, message = 'Request aborted before workflow mutation could be applied.' @@ -273,25 +323,31 @@ export async function executeRunWorkflow( const useDraftState = !params.useDeployedState const workflowInput = resolveRunWorkflowInput(params) - const result = await executeCopilotWorkflowUseCase(context, runWorkflowFromCopilot, { - workflowId, - assertedWorkspaceId: context.workspaceId, - useDraftState, - triggerBlockId: resolveRunTriggerBlockId(params), - workflowInput, - hasWorkflowInput: workflowInput !== undefined, - useMockPayload: params.useMockPayload === true, - inputFromExecutionId: resolveInputFromExecutionId(params.inputFromExecutionId), - lifecycle: copilotRunLifecycle(context), + return await answerWithinWait(context, workflowId, async (lifecycle) => { + try { + const result = await executeCopilotWorkflowUseCase(context, runWorkflowFromCopilot, { + workflowId, + assertedWorkspaceId: context.workspaceId, + useDraftState, + triggerBlockId: resolveRunTriggerBlockId(params), + workflowInput, + hasWorkflowInput: workflowInput !== undefined, + useMockPayload: params.useMockPayload === true, + inputFromExecutionId: resolveInputFromExecutionId(params.inputFromExecutionId), + lifecycle, + }) + + return buildExecutionOutput( + result, + context.resolvedSecretTraceRegistry, + settledPhase(result.status), + undefined, + params.select + ) + } catch (error) { + return buildExecutionError(error, context.resolvedSecretTraceRegistry) + } }) - - return buildExecutionOutput( - result, - context.resolvedSecretTraceRegistry, - settledPhase(result.status), - undefined, - params.select - ) } catch (error) { return buildExecutionError(error, context.resolvedSecretTraceRegistry) } @@ -351,26 +407,37 @@ export async function executeRunWorkflowUntilBlock( const useDraftState = !params.useDeployedState const workflowInput = resolveRunWorkflowInput(params) - const result = await executeCopilotWorkflowUseCase(context, runWorkflowUntilBlockFromCopilot, { - workflowId, - assertedWorkspaceId: context.workspaceId, - useDraftState, - triggerBlockId: resolveRunTriggerBlockId(params), - workflowInput, - hasWorkflowInput: workflowInput !== undefined, - useMockPayload: params.useMockPayload === true, - inputFromExecutionId: resolveInputFromExecutionId(params.inputFromExecutionId), - stopAfterBlockId: params.stopAfterBlockId, - lifecycle: copilotRunLifecycle(context), + const stopAfterBlockId = params.stopAfterBlockId + return await answerWithinWait(context, workflowId, async (lifecycle) => { + try { + const result = await executeCopilotWorkflowUseCase( + context, + runWorkflowUntilBlockFromCopilot, + { + workflowId, + assertedWorkspaceId: context.workspaceId, + useDraftState, + triggerBlockId: resolveRunTriggerBlockId(params), + workflowInput, + hasWorkflowInput: workflowInput !== undefined, + useMockPayload: params.useMockPayload === true, + inputFromExecutionId: resolveInputFromExecutionId(params.inputFromExecutionId), + stopAfterBlockId, + lifecycle, + } + ) + + return buildExecutionOutput( + result, + context.resolvedSecretTraceRegistry, + settledPhase(result.status), + { stoppedAfterBlockId: stopAfterBlockId }, + params.select + ) + } catch (error) { + return buildExecutionError(error, context.resolvedSecretTraceRegistry) + } }) - - return buildExecutionOutput( - result, - context.resolvedSecretTraceRegistry, - settledPhase(result.status), - { stoppedAfterBlockId: params.stopAfterBlockId }, - params.select - ) } catch (error) { return buildExecutionError(error, context.resolvedSecretTraceRegistry) } @@ -435,24 +502,31 @@ export async function executeRunFromBlock( } const useDraftState = !params.useDeployedState - const result = await executeCopilotWorkflowUseCase(context, runFromBlockFromCopilot, { - workflowId, - variableInputs: params.variableInputs, - assertedWorkspaceId: context.workspaceId, - useDraftState, - blockId: params.startBlockId, - workflowInput: resolveRunWorkflowInput(params), - sourceExecutionId: params.executionId, - lifecycle: copilotRunLifecycle(context), + const startBlockId = params.startBlockId + return await answerWithinWait(context, workflowId, async (lifecycle) => { + try { + const result = await executeCopilotWorkflowUseCase(context, runFromBlockFromCopilot, { + workflowId, + variableInputs: params.variableInputs, + assertedWorkspaceId: context.workspaceId, + useDraftState, + blockId: startBlockId, + workflowInput: resolveRunWorkflowInput(params), + sourceExecutionId: params.executionId, + lifecycle, + }) + + return buildExecutionOutput( + result, + context.resolvedSecretTraceRegistry, + settledPhase(result.status), + { startBlockId }, + params.select + ) + } catch (error) { + return buildExecutionError(error, context.resolvedSecretTraceRegistry) + } }) - - return buildExecutionOutput( - result, - context.resolvedSecretTraceRegistry, - settledPhase(result.status), - { startBlockId: params.startBlockId }, - params.select - ) } catch (error) { return buildExecutionError(error, context.resolvedSecretTraceRegistry) } @@ -472,24 +546,31 @@ export async function executeRunBlock( } const useDraftState = !params.useDeployedState - const result = await executeCopilotWorkflowUseCase(context, runBlockFromCopilot, { - workflowId, - variableInputs: params.variableInputs, - assertedWorkspaceId: context.workspaceId, - useDraftState, - blockId: params.blockId, - workflowInput: resolveRunWorkflowInput(params), - sourceExecutionId: params.executionId, - lifecycle: copilotRunLifecycle(context), + const blockId = params.blockId + return await answerWithinWait(context, workflowId, async (lifecycle) => { + try { + const result = await executeCopilotWorkflowUseCase(context, runBlockFromCopilot, { + workflowId, + variableInputs: params.variableInputs, + assertedWorkspaceId: context.workspaceId, + useDraftState, + blockId, + workflowInput: resolveRunWorkflowInput(params), + sourceExecutionId: params.executionId, + lifecycle, + }) + + return buildExecutionOutput( + result, + context.resolvedSecretTraceRegistry, + settledPhase(result.status), + { blockId }, + params.select + ) + } catch (error) { + return buildExecutionError(error, context.resolvedSecretTraceRegistry) + } }) - - return buildExecutionOutput( - result, - context.resolvedSecretTraceRegistry, - settledPhase(result.status), - { blockId: params.blockId }, - params.select - ) } catch (error) { return buildExecutionError(error, context.resolvedSecretTraceRegistry) } diff --git a/apps/sim/lib/mothership/tools/workflow-tools.ts b/apps/sim/lib/mothership/tools/workflow-tools.ts index 6858c9962a1..12dcebbddc5 100644 --- a/apps/sim/lib/mothership/tools/workflow-tools.ts +++ b/apps/sim/lib/mothership/tools/workflow-tools.ts @@ -187,6 +187,9 @@ export function createStructuralWorkflowToolCompletionData( } if (workflowId) data.workflowId = workflowId if (executionId) data.executionId = executionId + // A detached run is still going: the worker waits on it by these ids (WorkflowToolRunningResult). + if (status === ASYNC_TOOL_CONFIRMATION_STATUS.background && workflowId && executionId) + data.status = 'running' if (status === ASYNC_TOOL_CONFIRMATION_STATUS.error && launchError) { data.code = launchError.code data.error = launchError.message diff --git a/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts b/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts index 3a1fc488114..dcf07c96f1b 100644 --- a/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts +++ b/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts @@ -60,6 +60,11 @@ export interface CopilotWorkflowRunLifecycle { executionId: string copilotToolCallId: string } + /** + * The id the run takes when no execution is bound, chosen by a caller that must name the run + * before it settles (a workflow tool that answers a still-running run by its id). + */ + executionId?: string } interface BaseCopilotRunInput { @@ -268,7 +273,8 @@ async function executeCopilotRun(params: { const boundExecution = params.input.lifecycle.boundExecution // Reuse the caller's already-claimed execution id so the claim and the log // row describe the same run; otherwise mint our own as before. - const childExecutionId = boundExecution?.executionId ?? generateId() + const childExecutionId = + boundExecution?.executionId ?? params.input.lifecycle.executionId ?? generateId() const requestId = generateRequestId() const admission = await prepareWorkflowExecutionAdmission( { diff --git a/scripts/check-unused-exports.baseline.json b/scripts/check-unused-exports.baseline.json index 471af669731..aed26c01cc1 100644 --- a/scripts/check-unused-exports.baseline.json +++ b/scripts/check-unused-exports.baseline.json @@ -5070,7 +5070,6 @@ "apps/sim/lib/workflows/application/apply-workflow-operations.ts#WorkflowBlockEnabledChange", "apps/sim/lib/workflows/application/download-workflow-run-file.ts#DownloadWorkflowRunFileResult", "apps/sim/lib/workflows/application/resolve-workflow-outputs.ts#loadResolvedWorkflowOutputs", - "apps/sim/lib/workflows/application/run-workflow-from-copilot.ts#CopilotWorkflowRunLifecycle", "apps/sim/lib/workflows/application/update-workflow-content.ts#WorkflowVariableOperation", "apps/sim/lib/workflows/application/update-workflow.ts#AppliedWorkflowUpdate", "apps/sim/lib/workflows/application/update-workflow.ts#UpdateWorkflowPolicyInput", From 1b94af94142ae1f2f00a0a10eaae969d9a4e721b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 00:59:07 -0700 Subject: [PATCH 02/14] 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. --- .../mothership/async-runs/repository.test.ts | 50 ++++++++++++ .../lib/mothership/async-runs/repository.ts | 78 ++++++++++++++++++ .../request/application/stop-delivery.test.ts | 80 +++++++++++++++++++ .../request/application/stop-delivery.ts | 69 ++++++++++------ .../lib/organizations/member-manager.test.ts | 1 + .../src/mocks/mothership-async-runs.mock.ts | 6 +- 6 files changed, 256 insertions(+), 28 deletions(-) create mode 100644 apps/sim/lib/mothership/request/application/stop-delivery.test.ts diff --git a/apps/sim/lib/mothership/async-runs/repository.test.ts b/apps/sim/lib/mothership/async-runs/repository.test.ts index 9ef7c75bdde..ac34045b85e 100644 --- a/apps/sim/lib/mothership/async-runs/repository.test.ts +++ b/apps/sim/lib/mothership/async-runs/repository.test.ts @@ -14,6 +14,7 @@ import { closeStreamToolAdmission, createRunSegment, detachAsyncToolCall, + detachedWorkflowExecutionId, getClaimedWorkflowExecutionId, getUnsettledStreamSandboxProcesses, markAsyncToolRunning, @@ -458,3 +459,52 @@ describe('async tool repository single-row semantics', () => { ).rejects.toThrow('Async tool call belongs to another run') }) }) + +describe('the workflow run a settled call left running', () => { + const cli = (argv: string[], exitCode: number, record: Record) => ({ + executableName: 'sim_cli', + args: { request: { invocation: { kind: 'cli', argv } } }, + result: { exitCode, stdout: JSON.stringify(record), stderr: '' }, + }) + const running = { runId: 'run-1', workflowId: 'wf-1', status: 'running' } + + it('is a workflows run the CLI answered still running', () => { + expect( + detachedWorkflowExecutionId( + cli(['workflows', 'run', 'wf-1', '--wait-timeout', '240'], 4, running) + ) + ).toBe('run-1') + }) + + 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') + }) + + it.each([ + [ + 'a run that settled', + cli(['workflows', 'run', 'wf-1'], 0, { ...running, status: 'completed' }), + ], + // Waiting on a run proves nothing about who started it; Stop must not cancel it. + ['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' } }, + ], + ])('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 6383705b7b1..a6cb33a5b7a 100644 --- a/apps/sim/lib/mothership/async-runs/repository.ts +++ b/apps/sim/lib/mothership/async-runs/repository.ts @@ -261,6 +261,8 @@ export interface RunStopTarget { /** The client workflows and sandbox commands a stopped run started, read once its admission closed. */ export interface StoppedRunWork { clientWorkflowExecutionIds: string[] + /** Runs its calls started and left running past the call (`getDetachedWorkflowExecutions`). */ + detachedWorkflowExecutionIds: string[] sandboxProcesses: StoppedSandboxProcess[] } @@ -361,6 +363,7 @@ export async function stopLiveRunsTx( chatId, desktopDeviceId, clientWorkflowExecutionIds: await getUnsettledClientWorkflowExecutions(streamId, userId, tx), + detachedWorkflowExecutionIds: await getDetachedWorkflowExecutions(streamId, userId, tx), sandboxProcesses: await getUnsettledStreamSandboxProcesses(streamId, userId, tx), }) } @@ -1922,6 +1925,81 @@ 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 + * (`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 + if (call.executableName === 'sim_cli') { + const request = toRecordOrNull(toRecordOrNull(call.args)?.request) + const argv = toRecordOrNull(request?.invocation)?.argv + if (!Array.isArray(argv) || argv[0] !== 'workflows' || argv[1] !== 'run') return undefined + if (result.exitCode !== 4 || typeof result.stdout !== 'string') return undefined + try { + const record = toRecordOrNull(JSON.parse(result.stdout)) + return record?.status === 'running' && typeof record.runId === 'string' + ? record.runId + : undefined + } catch { + return undefined + } + } + return result.status === 'running' && typeof result.executionId === 'string' + ? result.executionId + : undefined +} + +/** + * 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. + */ +export async function getDetachedWorkflowExecutions( + streamId: string, + userId: string, + executor: DbOrTx = db +) { + return await withDbSpan( + TraceSpan.CopilotAsyncRunsGetRunSegment, + 'SELECT', + 'copilot_async_tool_calls', + { [TraceAttr.StreamId]: streamId, [TraceAttr.UserId]: userId }, + async () => { + const rows = await executor + .select({ + executableName: copilotAsyncToolCalls.executableName, + args: copilotAsyncToolCalls.args, + result: copilotAsyncToolCalls.result, + }) + .from(copilotAsyncToolCalls) + .innerJoin(copilotRuns, eq(copilotRuns.id, copilotAsyncToolCalls.runId)) + .where( + and( + eq(copilotRuns.streamId, streamId), + eq(copilotRuns.userId, userId), + inArray(copilotAsyncToolCalls.executableName, ['sim_cli', ...WORKFLOW_RUN_TOOL_NAMES]), + isNotNull(copilotAsyncToolCalls.result) + ) + ) + return [...new Set(rows.flatMap((row) => detachedWorkflowExecutionId(row) ?? []))] + } + ) +} + export function getClaimedWorkflowExecutionId(claimedBy: string | null | undefined) { if (!claimedBy?.startsWith(WORKFLOW_EXECUTION_CLAIM_PREFIX)) return undefined const executionId = claimedBy.slice(WORKFLOW_EXECUTION_CLAIM_PREFIX.length) diff --git a/apps/sim/lib/mothership/request/application/stop-delivery.test.ts b/apps/sim/lib/mothership/request/application/stop-delivery.test.ts new file mode 100644 index 00000000000..2c0bd4bd628 --- /dev/null +++ b/apps/sim/lib/mothership/request/application/stop-delivery.test.ts @@ -0,0 +1,80 @@ +import { + mothershipAsyncRunsMock, + mothershipAsyncRunsMockFns, +} from '@sim/testing/mocks/mothership-async-runs.mock' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const hoisted = vi.hoisted(() => ({ + abortManualExecution: vi.fn(), + markExecutionCancelled: vi.fn(), +})) +vi.mock('@/lib/mothership/async-runs/repository', () => mothershipAsyncRunsMock) +vi.mock('@/lib/execution/manual-cancellation', () => ({ + abortManualExecution: hoisted.abortManualExecution, +})) +vi.mock('@/lib/execution/cancellation', () => ({ + markExecutionCancelled: hoisted.markExecutionCancelled, +})) +vi.mock('@/lib/mothership/request/session/explicit-abort', () => ({ + requestExplicitStreamAbort: vi.fn(async () => ({ settled: true })), +})) +vi.mock('@/lib/mothership/request/session', () => ({ + abortActiveStream: vi.fn(), + waitForPendingChatStream: vi.fn(async () => true), + releasePendingChatStream: vi.fn(), +})) +vi.mock('@/lib/mothership/async-runs/orphaned-runs', () => ({ + settleStoppedRunWithoutController: vi.fn(async () => undefined), +})) +vi.mock('@/lib/mothership/request/tools/sandbox-recovery', () => ({ + recoverSandboxProcesses: vi.fn(async () => undefined), +})) +vi.mock('@/lib/desktop/executor/doorbell', () => ({ ringDesktopInbox: vi.fn() })) + +import { + deliverRunStop, + readStoppedRunWork, +} from '@/lib/mothership/request/application/stop-delivery' + +const run = { id: 'run-1', streamId: 'stream-1', userId: 'user-1', chatId: 'chat-1' } + +describe('a Stop', () => { + beforeEach(() => { + hoisted.abortManualExecution.mockReset() + hoisted.markExecutionCancelled.mockReset().mockResolvedValue({ durablyRecorded: true }) + }) + + // A synchronous run the worker waits on in slices has detached from its call: no call's abort + // signal reaches it, so before this a Stop mid-slice left the draft running to completion. + it('cancels the workflow runs its calls started and left running, beside its open client runs', async () => { + mothershipAsyncRunsMockFns.mockGetUnsettledClientWorkflowExecutions.mockResolvedValueOnce([ + 'client-1', + ]) + mothershipAsyncRunsMockFns.mockGetDetachedWorkflowExecutions.mockResolvedValueOnce([ + 'detached-1', + 'client-1', + ]) + const work = await readStoppedRunWork('stream-1', 'user-1') + expect(work).toEqual({ + clientWorkflowExecutionIds: ['client-1'], + detachedWorkflowExecutionIds: ['detached-1', 'client-1'], + sandboxProcesses: [], + }) + + await deliverRunStop(run, work) + + expect(hoisted.abortManualExecution.mock.calls).toEqual([['client-1'], ['detached-1']]) + expect(hoisted.markExecutionCancelled.mock.calls).toEqual([['client-1'], ['detached-1']]) + }) + + it('still cancels what it found when the detached runs cannot be read', async () => { + mothershipAsyncRunsMockFns.mockGetUnsettledClientWorkflowExecutions.mockResolvedValueOnce([ + 'client-1', + ]) + mothershipAsyncRunsMockFns.mockGetDetachedWorkflowExecutions.mockRejectedValueOnce( + new Error('db down') + ) + await deliverRunStop(run, await readStoppedRunWork('stream-1', 'user-1')) + expect(hoisted.markExecutionCancelled.mock.calls).toEqual([['client-1']]) + }) +}) diff --git a/apps/sim/lib/mothership/request/application/stop-delivery.ts b/apps/sim/lib/mothership/request/application/stop-delivery.ts index 76215fae5d5..cf43f5584ab 100644 --- a/apps/sim/lib/mothership/request/application/stop-delivery.ts +++ b/apps/sim/lib/mothership/request/application/stop-delivery.ts @@ -5,6 +5,7 @@ import { markExecutionCancelled } from '@/lib/execution/cancellation' import { abortManualExecution } from '@/lib/execution/manual-cancellation' import { settleStoppedRunWithoutController } from '@/lib/mothership/async-runs/orphaned-runs' import { + getDetachedWorkflowExecutions, getUnsettledClientWorkflowExecutions, getUnsettledStreamSandboxProcesses, type RunStopTarget, @@ -34,30 +35,39 @@ interface RunStopDeliveryOptions { } /** - * The client workflows and sandbox commands a stopped run started, read once its admission - * closed. Each read is best-effort: one that fails stops nothing it would have found. + * The client workflows, detached workflow runs and sandbox commands a stopped run started, read + * once its admission closed. Each read is best-effort: one that fails stops nothing it would have + * found. */ export async function readStoppedRunWork( streamId: string, userId: string ): Promise { - const [clientWorkflowExecutionIds, sandboxProcesses] = await Promise.all([ - getUnsettledClientWorkflowExecutions(streamId, userId).catch((error) => { - logger.warn('Stopped stream workflow ownership could not be read', { - streamId, - error: getErrorMessage(error), - }) - return [] - }), - getUnsettledStreamSandboxProcesses(streamId, userId).catch((error) => { - logger.warn('Stopped stream sandbox ownership could not be read', { - streamId, - error: getErrorMessage(error), - }) - return [] - }), - ]) - return { clientWorkflowExecutionIds, sandboxProcesses } + const [clientWorkflowExecutionIds, detachedWorkflowExecutionIds, sandboxProcesses] = + await Promise.all([ + getUnsettledClientWorkflowExecutions(streamId, userId).catch((error) => { + logger.warn('Stopped stream workflow ownership could not be read', { + streamId, + error: getErrorMessage(error), + }) + return [] + }), + getDetachedWorkflowExecutions(streamId, userId).catch((error) => { + logger.warn('Stopped stream detached workflow runs could not be read', { + streamId, + error: getErrorMessage(error), + }) + return [] + }), + getUnsettledStreamSandboxProcesses(streamId, userId).catch((error) => { + logger.warn('Stopped stream sandbox ownership could not be read', { + streamId, + error: getErrorMessage(error), + }) + return [] + }), + ]) + return { clientWorkflowExecutionIds, detachedWorkflowExecutionIds, sandboxProcesses } } /** @@ -97,14 +107,21 @@ export async function deliverRunStop( error: getErrorMessage(error), }) } - const cancelClientWorkflows = async () => - Promise.all( - (await work).clientWorkflowExecutionIds.map(async (executionId) => { - /** The execution route reserved these IDs before recording ownership; no client result is trusted. */ - abortManualExecution(executionId) - await markExecutionCancelled(executionId) - }) + const cancelClientWorkflows = async () => { + const { clientWorkflowExecutionIds, detachedWorkflowExecutionIds } = await work + /** + * The execution route reserved the client IDs before recording ownership; no client result is + * trusted. A detached run took no call's abort signal, so only its own cancel reaches it. + */ + return Promise.all( + [...new Set([...clientWorkflowExecutionIds, ...detachedWorkflowExecutionIds])].map( + async (executionId) => { + abortManualExecution(executionId) + await markExecutionCancelled(executionId) + } + ) ) + } const [settled, worker] = await Promise.all([ waitForPendingChatStream(chatId, 8000, streamId), workerStop, diff --git a/apps/sim/lib/organizations/member-manager.test.ts b/apps/sim/lib/organizations/member-manager.test.ts index 864f7581e2f..d8c6fe37ac0 100644 --- a/apps/sim/lib/organizations/member-manager.test.ts +++ b/apps/sim/lib/organizations/member-manager.test.ts @@ -107,6 +107,7 @@ describe('organization member managers', () => { chatId: 'chat', desktopDeviceId: null, clientWorkflowExecutionIds: ['execution'], + detachedWorkflowExecutionIds: [], sandboxProcesses: [], }, ] diff --git a/packages/testing/src/mocks/mothership-async-runs.mock.ts b/packages/testing/src/mocks/mothership-async-runs.mock.ts index cc6627ba539..36291d7fdf3 100644 --- a/packages/testing/src/mocks/mothership-async-runs.mock.ts +++ b/packages/testing/src/mocks/mothership-async-runs.mock.ts @@ -8,8 +8,8 @@ import { vi } from 'vitest' * its result (the advisory lock itself is not modelled). * - `mockGetRunSecretAuthority` resolves the no-secrets authority. * - `mockGetClaimedWorkflowExecutionId` is a faithful port (strips the real `'workflow:'` claim prefix). - * - `mockStopLiveRunsTx`, `mockFindWorkspacesWithLiveRunsTx`, `mockGetUnsettledStreamSandboxProcesses`, `mockGetUnsettledClientWorkflowExecutions` and - * `mockGetAsyncToolCalls` resolve `[]`. + * - `mockStopLiveRunsTx`, `mockFindWorkspacesWithLiveRunsTx`, `mockGetUnsettledStreamSandboxProcesses`, `mockGetUnsettledClientWorkflowExecutions`, + * `mockGetDetachedWorkflowExecutions` and `mockGetAsyncToolCalls` resolve `[]`. * * @example * ```ts @@ -64,6 +64,7 @@ export const mothershipAsyncRunsMockFns = { mockGetUnsettledClientWorkflowExecutions: vi.fn( async (..._args: unknown[]): Promise => [] ), + mockGetDetachedWorkflowExecutions: vi.fn(async (..._args: unknown[]): Promise => []), mockGetClaimedWorkflowExecutionId: vi.fn( (claimedBy: string | null | undefined): string | undefined => { if (!claimedBy?.startsWith('workflow:')) return undefined @@ -131,6 +132,7 @@ export const mothershipAsyncRunsMock = { areStreamToolExecutionsSettled: mothershipAsyncRunsMockFns.mockAreStreamToolExecutionsSettled, getUnsettledClientWorkflowExecutions: mothershipAsyncRunsMockFns.mockGetUnsettledClientWorkflowExecutions, + getDetachedWorkflowExecutions: mothershipAsyncRunsMockFns.mockGetDetachedWorkflowExecutions, getClaimedWorkflowExecutionId: mothershipAsyncRunsMockFns.mockGetClaimedWorkflowExecutionId, claimWorkflowToolExecution: mothershipAsyncRunsMockFns.mockClaimWorkflowToolExecution, releaseWorkflowToolExecutionClaim: From fa11739d399f0eb5c1ae626ef4542186e716b1f2 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 00:59:55 -0700 Subject: [PATCH 03/14] test(mothership): drive a workflow tool past its wait and a stop mid-slice through the real controller Two live cases beside the sliced draft run (opt-in, SIM_HELPERS_SMOKE and MSHIP_WORKER_ROOT): - tool: run_workflow on a Sim declaring workflow-tool-wait-timeout answers its 500 s run still running after 240 s, the worker waits on it as run-wait slices and reads it back, and the model sees the settled run's output. - declared-stop: the Stop button's request while the worker waits on a detached draft run ends the turn and cancels the run. The harness now marks the turn's tool execution and billing as the chat route does, registers and wires the turn's controller as lifecycle/start.ts does, gives the tool case a real AES key to seal its provenance, and keeps logged errors' messages in the evidence. --- .../agent-cli/saved-run-read.live.test.ts | 217 ++++++++++++++++-- 1 file changed, 204 insertions(+), 13 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 a835b935e38..862a9d3b36c 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 @@ -17,6 +17,7 @@ * Isolated columns plus required defaults/indexes do not prove migrations or all constraints. */ import { execFile, spawn } from 'node:child_process' +import { randomBytes } from 'node:crypto' import { once } from 'node:events' import { createReadStream } from 'node:fs' import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' @@ -62,7 +63,7 @@ import { workspaceFiles, } from '@sim/db/schema' import type { PermissionType } from '@sim/platform-authz/workspace' -import { authMockFns, envFlagsMock, setEnvFlags } from '@sim/testing' +import { authMockFns, envFlagsMock, setEnv, setEnvFlags } from '@sim/testing' import { sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import { eq, is, SQL, sql } from 'drizzle-orm' @@ -133,11 +134,18 @@ 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 { + WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + WORKFLOW_TOOL_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' -import { isToolCallStreamEvent } from '@/lib/mothership/request/session' +import { + isToolCallStreamEvent, + registerActiveStream, + unregisterActiveStream, +} from '@/lib/mothership/request/session' import { maybeWriteOutputToFile } from '@/lib/mothership/request/tools/files' import { maybeWriteOutputToTable } from '@/lib/mothership/request/tools/tables' import type { ExecutionContext } from '@/lib/mothership/request/types' @@ -154,6 +162,7 @@ import { calculateNextRunTime, getScheduleTimeValues } from '@/lib/workflows/sch import { fileOperations } from '@/lib/workspace-files/application/operations' import { readWorkspaceFileArtifact } from '@/lib/workspace-files/application/read-workspace-file-artifact' import { readWorkspaceFileText } from '@/lib/workspace-files/application/read-workspace-file-text' +import { POST as abortChatRoute } from '@/app/api/copilot/chat/abort/route' import { POST as addChatResourceRoute, DELETE as removeChatResourceRoute, @@ -230,7 +239,18 @@ vi.mock('@/lib/mothership/inbox/agentmail-client', () => ({ vi.mock('@sim/logger', async () => { const { loggerMock, createMockLogger } = await import('@sim/testing') const logger = createMockLogger() - logger.error.mockImplementation((...args: unknown[]) => fixture.errors.push(args)) + // Errors keep their message and top frames, which JSON evidence would otherwise drop as `{}`. + logger.error.mockImplementation((...args: unknown[]) => + fixture.errors.push( + JSON.parse( + JSON.stringify(args, (_key, value) => + value instanceof Error + ? { message: value.message, stack: value.stack?.split('\n').slice(0, 6) } + : value + ) + ) + ) + ) logger.withMetadata.mockReturnValue(logger) logger.child.mockReturnValue(logger) return { ...loggerMock, createLogger: () => logger } @@ -3601,6 +3621,10 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( ['undeclared', 260, 0], // The model's own --wait-timeout 5 lapses: an exit 4 the worker hands the model. ['own-budget', 30, 0], + // run_workflow on a Sim that answers a tool's run still running after the frame's wait. + ['tool', 300, 200], + // The user stops the chat while the worker waits on a detached run: the run stops too. + ['declared-stop', 300, 300], ] as const)( 'draft test run past a wait slice reaches the model settled (capability %s)', async (capability, waitA, waitB) => { @@ -3670,6 +3694,9 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( 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 + // A workflow tool call seals its inputs' provenance, which takes a real AES key. + const encryptionKey = env.ENCRYPTION_KEY + setEnv({ ENCRYPTION_KEY: randomBytes(32).toString('hex') }) const child = spawn('bun', ['tools/probes/src/draft-run-slices.ts'], { cwd: workerRoot, @@ -3686,6 +3713,7 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( MSHIP_PROBE_RUN_FLAGS: JSON.stringify( capability === 'own-budget' ? ['--wait-timeout', '5'] : [] ), + MSHIP_PROBE_TOOL: capability === 'tool' ? 'run_workflow' : '', }, stdio: ['pipe', 'pipe', 'pipe'], }) @@ -3734,8 +3762,51 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( workspaceId, type: 'mothership', title: `Long draft ${capability}`, + // The Stop route stops the chat's current stream. + ...(capability === 'declared-stop' ? { conversationId: messageId } : {}), }) const startedAt = Date.now() + /** When the declared-stop case stopped the chat, once the worker waited on a detached run. */ + let stoppedAt: number | undefined + // Registered as the chat route registers its turn (lifecycle/start.ts), so a Stop + // aborts it; the route also aborts it on an observed abort marker. + const turn = new AbortController() + registerActiveStream(messageId, turn) + const stopper = + capability === 'declared-stop' + ? (async () => { + // Bounded: a run that never reaches a run wait fails the case below. + for (let read = 0; read < 300; read++) { + await sleep(2_000) + const [current] = await db + .select({ id: copilotRuns.id }) + .from(copilotRuns) + .where(eq(copilotRuns.streamId, messageId)) + if (!current) continue + const waiting = await db + .select({ args: copilotAsyncToolCalls.args }) + .from(copilotAsyncToolCalls) + .where(eq(copilotAsyncToolCalls.runId, current.id)) + if (!JSON.stringify(waiting).includes('"runs","wait"')) continue + await sleep(5_000) + authMockFns.mockGetSession.mockResolvedValueOnce({ + user: { id: 'run-reader' }, + session: { id: 'stop-session' }, + }) + stoppedAt = Date.now() + // The Stop button's request: it stops the run, not just the stream. + const response = await abortChatRoute( + new NextRequest('https://sim.test/api/copilot/chat/abort', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ chatId, streamId: messageId, workspaceId }), + }) + ) + return response.status + } + return undefined + })() + : undefined const result = await runCopilotLifecycle( { message: 'Test-run the draft and tell me its output.', @@ -3745,9 +3816,16 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( chatId, messageId, // What lib/mothership/chat/payload.ts declares on every chat turn. - ...(capability !== 'undeclared' - ? { clientCapabilities: [WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY] } - : {}), + ...(capability === 'tool' + ? { + clientCapabilities: [ + WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY, + WORKFLOW_TOOL_WAIT_TIMEOUT_CAPABILITY, + ], + } + : capability !== 'undeclared' + ? { clientCapabilities: [WORKFLOW_RUN_WAIT_TIMEOUT_CAPABILITY] } + : {}), }, { userId: 'run-reader', @@ -3758,12 +3836,32 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( interactive: false, clientToolPickupExpected: false, flushAfterEvent: false, - abortSignal: AbortSignal.timeout(20 * 60_000), + // Wired as the chat route wires its controller (lifecycle/start.ts): an observed + // Stop aborts the turn. + abortSignal: AbortSignal.any([turn.signal, AbortSignal.timeout(20 * 60_000)]), + onAbortObserved: (reason) => { + if (!turn.signal.aborted) turn.abort(reason) + }, executionContext: { userId: 'run-reader', workspaceId, workflowId: draftId, chatId, + // As the chat route marks every turn's tool execution and bills it + // (lib/mothership/chat/post.ts); a workflow tool runs in-process under both. + copilotToolExecution: true, + billingAttribution: { + actorUserId: 'run-reader', + workspaceId, + organizationId: null, + billedAccountUserId: 'run-reader', + billingEntity: { type: 'user', id: 'run-reader' }, + billingPeriod: { + start: '2026-09-01T00:00:00.000Z', + end: '2026-10-01T00:00:00.000Z', + }, + payerSubscription: null, + }, userPermission: 'write', resolvedSecretTraceRegistry: new ResolvedSecretTraceRegistry([], { userId: 'run-reader', @@ -3773,6 +3871,8 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( } ) const endedAt = Date.now() + unregisterActiveStream(messageId, turn) + const stopStatus = await stopper const [run] = await db .select() .from(copilotRuns) @@ -3783,8 +3883,38 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( .from(copilotAsyncToolCalls) .where(eq(copilotAsyncToolCalls.runId, run.id)) : [] + /** The detached run's own status after a Stop, read until it leaves `running`. */ + let stoppedRun: { executionId?: string; status?: string; endedAt?: string } = {} + if (capability === 'declared-stop') { + const receipt = calls.find( + (call) => + (call.args as { request?: { invocation?: { argv?: string[] } } }).request + ?.invocation?.argv?.[1] === 'run' + )?.result as { stdout?: string } | undefined + const executionId = String(JSON.parse(receipt?.stdout ?? '{}').runId ?? '') + for (let read = 0; read < 30; read++) { + const [log] = await db + .select({ + status: workflowExecutionLogs.status, + endedAt: workflowExecutionLogs.endedAt, + }) + .from(workflowExecutionLogs) + .where(eq(workflowExecutionLogs.executionId, executionId)) + stoppedRun = { + executionId, + status: log?.status, + endedAt: log?.endedAt?.toISOString(), + } + if (log && log.status !== 'running') break + await sleep(2_000) + } + } const evidence = { capability, + stopStatus, + stoppedAt: stoppedAt ? new Date(stoppedAt).toISOString() : undefined, + stoppedRun, + errors: fixture.errors, waits: [waitA, waitB], startedAt: new Date(startedAt).toISOString(), totalSeconds: (endedAt - startedAt) / 1000, @@ -3798,6 +3928,9 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( toolName: call.toolName, args: call.args, status: call.status, + createdAt: call.createdAt?.toISOString(), + result: call.toolName === 'sim_cli' ? undefined : call.result, + error: call.error, exitCode: (call.result as { exitCode?: unknown } | null)?.exitCode, stdout: (call.result as { stdout?: unknown } | null)?.stdout, startedAt: call.executionStartedAt?.toISOString(), @@ -3818,13 +3951,24 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( `${process.env.MSHIP_SLICES_REPORT_PATH}.${capability}.json`, JSON.stringify(evidence, null, 2) ) + if (capability === 'declared-stop') { + expect(stopStatus).toBe(200) + // The turn ended at the Stop, not when the run would have settled. + expect(endedAt - (stoppedAt ?? endedAt)).toBeLessThan(60_000) + // The run the worker was waiting on stopped with the chat, as a held run does. + expect(stoppedRun.status, JSON.stringify(evidence)).toBe('cancelled') + return + } 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) + const seen = + typeof report.output.value === 'string' + ? report.output.value + : JSON.stringify(report.output.value) if (capability === 'own-budget') { expect(calls).toHaveLength(1) expect( @@ -3840,10 +3984,6 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( } // 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 @@ -3851,6 +3991,56 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( call.result as { exitCode: number; stdout: string } const heldSeconds = (call: (typeof calls)[number]) => (Number(call.executionSettledAt) - Number(call.executionStartedAt)) / 1000 + if (capability === 'tool') { + const [toolCall, ...slices] = calls + expect(toolCall.toolName).toBe('run_workflow') + // The tool answered its run still running by the run's id, after one wait. + const running = toolCall.result as { status?: string; executionId?: string } + expect(running).toMatchObject({ status: 'running', workflowId: draftId }) + const executionId = String(running.executionId) + const answeredAfter = (Number(slices[0].createdAt) - Number(toolCall.createdAt)) / 1000 + expect(answeredAfter).toBeGreaterThan(239) + expect(answeredAfter).toBeLessThan(250) + const toolWaits = slices.filter( + (call) => argvOf(call).slice(0, 3).join(' ') === 'workflows runs wait' + ) + expect(toolWaits.length).toBeGreaterThan(0) + for (const [index, call] of toolWaits.entries()) { + expect(argvOf(call)).toEqual([ + 'workflows', + 'runs', + 'wait', + executionId, + '--workflow', + draftId, + '--wait-timeout', + '240', + ]) + expect(outcome(call).exitCode).toBe(index === toolWaits.length - 1 ? 0 : 4) + expect(heldSeconds(call)).toBeLessThan(250) + } + const readBack = slices.at(-1) + if (!readBack) throw new Error('The run was never read back') + expect(argvOf(readBack)).toEqual([ + 'workflows', + 'runs', + 'get', + executionId, + '--workflow', + draftId, + '--include-output', + ]) + expect(slices).toHaveLength(toolWaits.length + 1) + // The model saw the settled run with its output, not the still-running answer. + expect(seen).toContain('"status":"completed"') + expect(seen).toContain('after-waits') + expect(endedAt - startedAt).toBeGreaterThan(waitSeconds * 1000) + return + } + // 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 runs = calls.filter((call) => argvOf(call)[1] === 'run') const waits = calls.filter( (call) => argvOf(call).slice(0, 3).join(' ') === 'workflows runs wait' @@ -3919,6 +4109,7 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( expect(endedAt - startedAt).toBeGreaterThan(waitSeconds * 1000) } finally { clearTimeout(startupTimeout) + setEnv({ ENCRYPTION_KEY: encryptionKey }) fixture.executionTimeoutMs = 20_000 child.kill() await exited From 7543bf3ae8acefd5a982e6be41e806e5fac9f21e Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:12:31 -0700 Subject: [PATCH 04/14] test(mothership): type the stopped run fixture and keep logged errors without a JSON round trip --- .../agent-cli/saved-run-read.live.test.ts | 20 +++++++++---------- .../request/application/stop-delivery.test.ts | 8 +++++++- 2 files changed, 16 insertions(+), 12 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 862a9d3b36c..5b96af7be0a 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 @@ -240,17 +240,15 @@ vi.mock('@sim/logger', async () => { const { loggerMock, createMockLogger } = await import('@sim/testing') const logger = createMockLogger() // Errors keep their message and top frames, which JSON evidence would otherwise drop as `{}`. - logger.error.mockImplementation((...args: unknown[]) => - fixture.errors.push( - JSON.parse( - JSON.stringify(args, (_key, value) => - value instanceof Error - ? { message: value.message, stack: value.stack?.split('\n').slice(0, 6) } - : value - ) - ) - ) - ) + const loggable = (value: unknown): unknown => + value instanceof Error + ? { message: value.message, stack: value.stack?.split('\n').slice(0, 6) } + : Array.isArray(value) + ? value.map(loggable) + : value !== null && typeof value === 'object' + ? Object.fromEntries(Object.entries(value).map(([key, item]) => [key, loggable(item)])) + : value + logger.error.mockImplementation((...args: unknown[]) => fixture.errors.push(args.map(loggable))) logger.withMetadata.mockReturnValue(logger) logger.child.mockReturnValue(logger) return { ...loggerMock, createLogger: () => logger } diff --git a/apps/sim/lib/mothership/request/application/stop-delivery.test.ts b/apps/sim/lib/mothership/request/application/stop-delivery.test.ts index 2c0bd4bd628..f5fd99706e7 100644 --- a/apps/sim/lib/mothership/request/application/stop-delivery.test.ts +++ b/apps/sim/lib/mothership/request/application/stop-delivery.test.ts @@ -36,7 +36,13 @@ import { readStoppedRunWork, } from '@/lib/mothership/request/application/stop-delivery' -const run = { id: 'run-1', streamId: 'stream-1', userId: 'user-1', chatId: 'chat-1' } +const run = { + id: 'run-1', + streamId: 'stream-1', + userId: 'user-1', + chatId: 'chat-1', + desktopDeviceId: null, +} describe('a Stop', () => { beforeEach(() => { From 3a8112d7f93517c7654a3742518785daf91cc4e5 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:20:22 -0700 Subject: [PATCH 05/14] fix(mothership): answer a workflow tool still running only once its run is dispatched Review of the wait: - A run still in authorization or admission when the wait passes is not answered running under an id nothing has used yet; the call waits until the run is handed to the executor (onDispatch) or settles. - The browser-run detacher sleeps interruptibly and stops on either outcome of its waiter, so no derived promise rejects unhandled. - The live case takes its AES key from generateRandomHex. - The mock-only Stop delivery test is dropped: the live declared-stop case drives the real Stop route, storage and cancellation, and fails without the fix (the run stays running); detachedWorkflowExecutionId keeps its unit test. --- .../agent-cli/saved-run-read.live.test.ts | 4 +- .../request/application/stop-delivery.test.ts | 86 ------------------- .../request/tools/workflow-client-fallback.ts | 7 +- .../tools/handlers/workflow/mutations.test.ts | 25 +++++- .../tools/handlers/workflow/mutations.ts | 9 +- .../application/run-workflow-from-copilot.ts | 3 + 6 files changed, 40 insertions(+), 94 deletions(-) delete mode 100644 apps/sim/lib/mothership/request/application/stop-delivery.test.ts 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 5b96af7be0a..f92196fcedc 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 @@ -17,7 +17,6 @@ * Isolated columns plus required defaults/indexes do not prove migrations or all constraints. */ import { execFile, spawn } from 'node:child_process' -import { randomBytes } from 'node:crypto' import { once } from 'node:events' import { createReadStream } from 'node:fs' import { mkdir, mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' @@ -66,6 +65,7 @@ import type { PermissionType } from '@sim/platform-authz/workspace' import { authMockFns, envFlagsMock, setEnv, setEnvFlags } from '@sim/testing' import { sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' +import { generateRandomHex } from '@sim/utils/random' import { eq, is, SQL, sql } from 'drizzle-orm' import { getTableConfig } from 'drizzle-orm/pg-core' import { NextRequest } from 'next/server' @@ -3694,7 +3694,7 @@ describe.skipIf(!process.env.MSHIP_TEST_DATABASE_URL)( fixture.executionTimeoutMs = 50 * 60_000 // A workflow tool call seals its inputs' provenance, which takes a real AES key. const encryptionKey = env.ENCRYPTION_KEY - setEnv({ ENCRYPTION_KEY: randomBytes(32).toString('hex') }) + setEnv({ ENCRYPTION_KEY: generateRandomHex(64) }) const child = spawn('bun', ['tools/probes/src/draft-run-slices.ts'], { cwd: workerRoot, diff --git a/apps/sim/lib/mothership/request/application/stop-delivery.test.ts b/apps/sim/lib/mothership/request/application/stop-delivery.test.ts deleted file mode 100644 index f5fd99706e7..00000000000 --- a/apps/sim/lib/mothership/request/application/stop-delivery.test.ts +++ /dev/null @@ -1,86 +0,0 @@ -import { - mothershipAsyncRunsMock, - mothershipAsyncRunsMockFns, -} from '@sim/testing/mocks/mothership-async-runs.mock' -import { beforeEach, describe, expect, it, vi } from 'vitest' - -const hoisted = vi.hoisted(() => ({ - abortManualExecution: vi.fn(), - markExecutionCancelled: vi.fn(), -})) -vi.mock('@/lib/mothership/async-runs/repository', () => mothershipAsyncRunsMock) -vi.mock('@/lib/execution/manual-cancellation', () => ({ - abortManualExecution: hoisted.abortManualExecution, -})) -vi.mock('@/lib/execution/cancellation', () => ({ - markExecutionCancelled: hoisted.markExecutionCancelled, -})) -vi.mock('@/lib/mothership/request/session/explicit-abort', () => ({ - requestExplicitStreamAbort: vi.fn(async () => ({ settled: true })), -})) -vi.mock('@/lib/mothership/request/session', () => ({ - abortActiveStream: vi.fn(), - waitForPendingChatStream: vi.fn(async () => true), - releasePendingChatStream: vi.fn(), -})) -vi.mock('@/lib/mothership/async-runs/orphaned-runs', () => ({ - settleStoppedRunWithoutController: vi.fn(async () => undefined), -})) -vi.mock('@/lib/mothership/request/tools/sandbox-recovery', () => ({ - recoverSandboxProcesses: vi.fn(async () => undefined), -})) -vi.mock('@/lib/desktop/executor/doorbell', () => ({ ringDesktopInbox: vi.fn() })) - -import { - deliverRunStop, - readStoppedRunWork, -} from '@/lib/mothership/request/application/stop-delivery' - -const run = { - id: 'run-1', - streamId: 'stream-1', - userId: 'user-1', - chatId: 'chat-1', - desktopDeviceId: null, -} - -describe('a Stop', () => { - beforeEach(() => { - hoisted.abortManualExecution.mockReset() - hoisted.markExecutionCancelled.mockReset().mockResolvedValue({ durablyRecorded: true }) - }) - - // A synchronous run the worker waits on in slices has detached from its call: no call's abort - // signal reaches it, so before this a Stop mid-slice left the draft running to completion. - it('cancels the workflow runs its calls started and left running, beside its open client runs', async () => { - mothershipAsyncRunsMockFns.mockGetUnsettledClientWorkflowExecutions.mockResolvedValueOnce([ - 'client-1', - ]) - mothershipAsyncRunsMockFns.mockGetDetachedWorkflowExecutions.mockResolvedValueOnce([ - 'detached-1', - 'client-1', - ]) - const work = await readStoppedRunWork('stream-1', 'user-1') - expect(work).toEqual({ - clientWorkflowExecutionIds: ['client-1'], - detachedWorkflowExecutionIds: ['detached-1', 'client-1'], - sandboxProcesses: [], - }) - - await deliverRunStop(run, work) - - expect(hoisted.abortManualExecution.mock.calls).toEqual([['client-1'], ['detached-1']]) - expect(hoisted.markExecutionCancelled.mock.calls).toEqual([['client-1'], ['detached-1']]) - }) - - it('still cancels what it found when the detached runs cannot be read', async () => { - mothershipAsyncRunsMockFns.mockGetUnsettledClientWorkflowExecutions.mockResolvedValueOnce([ - 'client-1', - ]) - mothershipAsyncRunsMockFns.mockGetDetachedWorkflowExecutions.mockRejectedValueOnce( - new Error('db down') - ) - await deliverRunStop(run, await readStoppedRunWork('stream-1', 'user-1')) - expect(hoisted.markExecutionCancelled.mock.calls).toEqual([['client-1']]) - }) -}) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts index 0d2dac3eb47..dee7b3ec7c1 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts @@ -1,6 +1,6 @@ import { createLogger } from '@sim/logger' import { toError } from '@sim/utils/errors' -import { sleep } from '@sim/utils/helpers' +import { interruptibleSleep, sleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import type { AsyncCompletionSignal, @@ -79,7 +79,7 @@ async function detachClientRunAfterWait(params: { if (!workflowId) return let delayMs = params.waitTimeoutMs while (!signal.aborted) { - await sleep(delayMs) + await interruptibleSleep(delayMs, signal) delayMs = CLIENT_CLAIM_POLL_MS if (signal.aborted) return const row = await getAsyncToolCall(toolCallId) @@ -140,7 +140,8 @@ export async function raceWorkflowToolClientPickup( }) if (params.waitTimeoutMs !== undefined) { const stopDetach = new AbortController() - void clientWait.finally(() => stopDetach.abort()) + const stop = () => stopDetach.abort() + void clientWait.then(stop, stop) detachClientRunAfterWait({ toolCallId, workflowId, 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 9c5647a78ca..e2c83f9cd48 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts @@ -440,12 +440,17 @@ describe('a workflow tool run past its wait', () => { }) /** A run the test settles itself, and the lifecycle it was started with. */ - function pendingRun() { + function pendingRun(dispatch = true) { let settle: (value: unknown) => void = () => {} const started = new Promise((resolve) => { settle = resolve }) - mocks.executeWorkflowUseCase.mockReturnValueOnce(started) + mocks.executeWorkflowUseCase.mockImplementationOnce( + (_context: unknown, _useCase: unknown, input: { lifecycle: { onDispatch?: () => void } }) => { + if (dispatch) input.lifecycle.onDispatch?.() + return started + } + ) return { settle, lifecycle: () => @@ -488,6 +493,22 @@ describe('a workflow tool run past its wait', () => { await expect(held[0]).resolves.toMatchObject({ success: true }) }) + it('answers a run still in preflight past its wait with how it settles', async () => { + const run = pendingRun(false) + let answered = false + const answer = executeRunWorkflow( + { workflowId: 'workflow-1' }, + { ...context, workflowRunWaitMs: 1_000 } + ).then((result) => { + answered = true + return result + }) + await vi.advanceTimersByTimeAsync(5_000) + expect(answered).toBe(false) + run.settle({ success: false, error: 'Workflow not found', logs: [] }) + await expect(answer).resolves.toMatchObject({ success: false }) + }) + it('answers with the result when the run settles within its wait', async () => { const run = pendingRun() const answer = executeRunBlock( diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts index 08da7649253..b1391108e52 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts @@ -271,7 +271,11 @@ async function answerWithinWait( const waitMs = context.workflowRunWaitMs if (waitMs === undefined) return run(lifecycle) const executionId = lifecycle.boundExecution?.executionId ?? generateId() - const settled = run({ ...lifecycle, executionId }) + let markDispatched: () => void = () => {} + const dispatched = new Promise((resolve) => { + markDispatched = () => resolve(null) + }) + const settled = run({ ...lifecycle, executionId, onDispatch: () => markDispatched() }) retainSandboxExecution(settled) let timer: ReturnType | undefined const waited = new Promise((resolve) => { @@ -280,6 +284,9 @@ async function answerWithinWait( try { const result = await Promise.race([settled, waited]) if (result) return result + // Only a run handed to the executor is answered running; one still in preflight settles first. + const started = await Promise.race([settled, dispatched]) + if (started) return started } finally { clearTimeout(timer) } diff --git a/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts b/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts index dcf07c96f1b..080cc0308ec 100644 --- a/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts +++ b/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts @@ -65,6 +65,8 @@ export interface CopilotWorkflowRunLifecycle { * before it settles (a workflow tool that answers a still-running run by its id). */ executionId?: string + /** Called once authorization and admission passed and the run is handed to the executor. */ + onDispatch?: () => void } interface BaseCopilotRunInput { @@ -307,6 +309,7 @@ async function executeCopilotRun(params: { * can already make with the id it was handed. Keep the executor call first: anything * inserted above it would be reported as a run that may exist. */ + params.input.lifecycle.onDispatch?.() try { const result = await executeWorkflow( { From b9e89bae869760e5356516a15409b899cfef9a1a Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:27:39 -0700 Subject: [PATCH 06/14] test(mothership): prove the browser detach and the Stop read against real Postgres and Redis The browser-run detach and the detached-run read now run through the real claim, settlement, published confirmation, waiter and stored tool rows (workflow-client-settlement.integration.ts), replacing unit tests that asserted only on mocks. --- .../tools/workflow-client-fallback.test.ts | 92 +------------------ .../workflow-client-settlement.integration.ts | 78 +++++++++++++++- 2 files changed, 78 insertions(+), 92 deletions(-) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.test.ts b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.test.ts index 13e50f4c49d..ec29de68d56 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.test.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.test.ts @@ -8,13 +8,8 @@ import { } from '@sim/testing/mocks/mothership-client-tool-waiter.mock' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' -const { recordDegraded, settleClientToolCall } = vi.hoisted(() => ({ +const { recordDegraded } = vi.hoisted(() => ({ recordDegraded: vi.fn(), - settleClientToolCall: vi.fn(), -})) - -vi.mock('@/lib/mothership/request/tools/client-settlement.server', () => ({ - settleClientToolCall, })) vi.mock('@/lib/mothership/request/metrics', () => ({ recordDegraded })) @@ -29,8 +24,6 @@ vi.mock('@/lib/mothership/async-runs/repository', () => mothershipAsyncRunsMock) import { raceWorkflowToolClientPickup } from '@/lib/mothership/request/tools/workflow-client-fallback' const claimWorkflowToolExecution = mothershipAsyncRunsMockFns.mockClaimWorkflowToolExecution -const getAsyncToolCall = mothershipAsyncRunsMockFns.mockGetAsyncToolCall -const getClaimedWorkflowExecutionId = mothershipAsyncRunsMockFns.mockGetClaimedWorkflowExecutionId const GRACE_MS = 30_000 const TIMEOUT_MS = 3_600_000 @@ -223,86 +216,3 @@ describe('raceWorkflowToolClientPickup', () => { expect(params.runOnServer).not.toHaveBeenCalled() }) }) - -describe('a browser-run workflow call past its wait', () => { - const WAIT_MS = 240_000 - beforeEach(() => { - vi.useFakeTimers() - waiterSignals = [] - settleClientToolCall.mockReset() - getAsyncToolCall.mockReset() - getClaimedWorkflowExecutionId.mockImplementation((claimedBy?: string | null) => - claimedBy?.startsWith('workflow:') ? claimedBy.slice('workflow:'.length) : undefined - ) - pendingUntilAborted() - // A browser already holds the call, so the server keeps waiting on it. - claimWorkflowToolExecution.mockResolvedValue(null) - }) - afterEach(() => { - vi.useRealTimers() - }) - - const browserClaim = { - status: 'running', - claimedBy: 'workflow:execution-1', - clientWorkflowExecutionId: 'execution-1', - } - - it('settles the call running in the background by the browser run ids', async () => { - getAsyncToolCall.mockResolvedValue(browserClaim) - const params = baseParams({ waitTimeoutMs: WAIT_MS }) - void raceWorkflowToolClientPickup(params as never) - - await vi.advanceTimersByTimeAsync(WAIT_MS - 1) - expect(settleClientToolCall).not.toHaveBeenCalled() - await vi.advanceTimersByTimeAsync(1) - - expect(settleClientToolCall).toHaveBeenCalledWith({ - toolCallId: 'tool-1', - status: 'background', - message: 'Workflow execution is continuing in the background.', - data: { workflowId: 'workflow-1', executionId: 'execution-1', status: 'running' }, - executionId: 'execution-1', - guard: { kind: 'open' }, - }) - // The browser's run goes on; only the server's wait ended. - expect(params.runOnServer).not.toHaveBeenCalled() - }) - - it('re-reads a call no browser has claimed yet until one does', async () => { - getAsyncToolCall - .mockResolvedValueOnce({ - status: 'running', - claimedBy: null, - clientWorkflowExecutionId: null, - }) - .mockResolvedValue(browserClaim) - void raceWorkflowToolClientPickup(baseParams({ waitTimeoutMs: WAIT_MS }) as never) - - await vi.advanceTimersByTimeAsync(WAIT_MS) - expect(settleClientToolCall).not.toHaveBeenCalled() - await vi.advanceTimersByTimeAsync(5_000) - expect(settleClientToolCall).toHaveBeenCalledWith( - expect.objectContaining({ status: 'background', executionId: 'execution-1' }) - ) - }) - - it('leaves a call that settled first alone', async () => { - waitForWorkflowToolCompletion.mockResolvedValue({ status: 'success', data: { ok: true } }) - const outcome = await raceWorkflowToolClientPickup( - baseParams({ waitTimeoutMs: WAIT_MS }) as never - ) - await vi.advanceTimersByTimeAsync(WAIT_MS * 2) - expect(outcome.winner).toBe('client') - expect(getAsyncToolCall).not.toHaveBeenCalled() - expect(settleClientToolCall).not.toHaveBeenCalled() - }) - - it('holds a call that names no wait', async () => { - getAsyncToolCall.mockResolvedValue(browserClaim) - void raceWorkflowToolClientPickup(baseParams() as never) - await vi.advanceTimersByTimeAsync(TIMEOUT_MS / 2) - expect(getAsyncToolCall).not.toHaveBeenCalled() - expect(settleClientToolCall).not.toHaveBeenCalled() - }) -}) 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 f5406b717c3..e730df540f6 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 @@ -36,10 +36,12 @@ import { claimWorkflowToolExecution, completeAsyncToolCall, detachAsyncToolCall, + getDetachedWorkflowExecutions, settleClientWorkflowToolExecution, } from '@/lib/mothership/async-runs/repository' import { getToolConfirmation } from '@/lib/mothership/persistence/tool-confirm' import { waitForWorkflowToolCompletion } from '@/lib/mothership/request/tools/client' +import { raceWorkflowToolClientPickup } from '@/lib/mothership/request/tools/workflow-client-fallback' import { reportQueuedClientWorkflowTool, reportSettledClientWorkflowTool, @@ -88,6 +90,7 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => const workflowId = generateId() const chatId = generateId() const runId = generateId() + const streamId = generateId() const snapshotIds: string[] = [] beforeAll(async () => { @@ -128,7 +131,7 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => chatId, userId, workspaceId, - streamId: generateId(), + streamId, toolExecutionVersion: SIM_TOOL_EXECUTION_VERSION, status: 'paused_waiting_for_tool', requestContext: { source: 'headless_lifecycle' }, @@ -490,4 +493,77 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => }) expect(await publishedConfirmation(toolCallId)).toBeNull() }) + + it("settles a browser run that outlasts the worker's wait as running, bound to the call", async () => { + const { toolCallId, executionId } = await claimed() + const runOnServer = vi.fn() + + const outcome = await raceWorkflowToolClientPickup({ + toolCallId, + workflowId, + timeoutMs: WAIT_MS, + graceMs: WAIT_MS, + waitTimeoutMs: 500, + runOnServer, + }) + + // The worker gets the run's ids and waits on it in slices; the browser's run goes on. + expect(outcome).toMatchObject({ + winner: 'client', + completion: { + status: 'background', + data: { workflowId, executionId, status: 'running' }, + }, + }) + expect(runOnServer).not.toHaveBeenCalled() + expect(await publishedConfirmation(toolCallId)).toMatchObject({ + status: 'background', + executionId, + }) + // Still bound to the call and unsettled, so a Stop's open-client-run read still cancels it. + expect(await toolRow(toolCallId)).toMatchObject({ + status: 'delivered', + claimedBy: `workflow:${executionId}`, + clientWorkflowExecutionId: executionId, + executionSettledAt: null, + }) + }) + + 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({ + runId, + toolCallId: generateId(), + toolName: executableName, + executableName, + args: args as Record, + result: result as Record, + status: 'completed', + }) + const detachedCli = generateId() + const detachedTool = generateId() + 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: '' } + ) + // A wait on a run proves nothing about who started it. + await insertCall( + 'sim_cli', + { request: { invocation: { kind: 'cli', argv: ['workflows', 'runs', 'wait', 'other'] } } }, + { exitCode: 4, stdout: record(generateId(), 'running'), stderr: '' } + ) + await insertCall( + 'run_workflow', + { workflowId }, + { status: 'running', executionId: detachedTool, workflowId } + ) + await insertCall('run_block', { workflowId }, { success: true, executionId: generateId() }) + + expect((await getDetachedWorkflowExecutions(streamId, userId)).sort()).toEqual( + [detachedCli, detachedTool].sort() + ) + }) }) From 04886f78dfcd54377061584b27a73a017546104c Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:32:02 -0700 Subject: [PATCH 07/14] refactor(mothership): wait on a workflow tool's run with the shared interruptible sleep --- .../lib/mothership/tools/handlers/workflow/mutations.ts | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts index b1391108e52..0d92b8557fc 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts @@ -1,4 +1,5 @@ import { createLogger } from '@sim/logger' +import { interruptibleSleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import { filterUndefined, isPlainRecord, isRecordLike } from '@sim/utils/object' import { createCopilotWorkspaceApiKey } from '@/lib/api-key/application/create-api-key' @@ -277,10 +278,8 @@ async function answerWithinWait( }) const settled = run({ ...lifecycle, executionId, onDispatch: () => markDispatched() }) retainSandboxExecution(settled) - let timer: ReturnType | undefined - const waited = new Promise((resolve) => { - timer = setTimeout(() => resolve(null), waitMs) - }) + const stopWaiting = new AbortController() + const waited = interruptibleSleep(waitMs, stopWaiting.signal).then(() => null) try { const result = await Promise.race([settled, waited]) if (result) return result @@ -288,7 +287,7 @@ async function answerWithinWait( const started = await Promise.race([settled, dispatched]) if (started) return started } finally { - clearTimeout(timer) + stopWaiting.abort() } logger.info('Workflow tool run outlasted its wait; answering it still running', { toolCallId: context.toolCallId, From 3cc2d8dca0a9c2eae6e3222b69d14353251ac97b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:35:35 -0700 Subject: [PATCH 08/14] fix(mothership): find a detached workflows run behind global CLI flags --- apps/sim/lib/mothership/async-runs/repository.test.ts | 8 ++++++++ apps/sim/lib/mothership/async-runs/repository.ts | 6 +++++- 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/apps/sim/lib/mothership/async-runs/repository.test.ts b/apps/sim/lib/mothership/async-runs/repository.test.ts index ac34045b85e..b31c12226c0 100644 --- a/apps/sim/lib/mothership/async-runs/repository.test.ts +++ b/apps/sim/lib/mothership/async-runs/repository.test.ts @@ -476,6 +476,14 @@ describe('the workflow run a settled call left running', () => { ).toBe('run-1') }) + it('is a workflows run behind global flags', () => { + expect( + detachedWorkflowExecutionId( + cli(['--output', 'json', 'workflows', 'run', 'wf-1', '--wait-timeout', '240'], 4, running) + ) + ).toBe('run-1') + }) + it('is a workflow tool answered still running', () => { expect( detachedWorkflowExecutionId({ diff --git a/apps/sim/lib/mothership/async-runs/repository.ts b/apps/sim/lib/mothership/async-runs/repository.ts index a6cb33a5b7a..bc441b9165f 100644 --- a/apps/sim/lib/mothership/async-runs/repository.ts +++ b/apps/sim/lib/mothership/async-runs/repository.ts @@ -1948,7 +1948,11 @@ export function detachedWorkflowExecutionId(call: { if (call.executableName === 'sim_cli') { const request = toRecordOrNull(toRecordOrNull(call.args)?.request) const argv = toRecordOrNull(request?.invocation)?.argv - if (!Array.isArray(argv) || argv[0] !== 'workflows' || argv[1] !== 'run') return undefined + // 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 try { const record = toRecordOrNull(JSON.parse(result.stdout)) From 6d2db309889cb3ad0c9c37eb533a8a9dd32b2292 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:40:40 -0700 Subject: [PATCH 09/14] fix(mothership): answer a workflow tool still running only once its run is readable A run past the call's wait was answered running as soon as it was handed to the executor, before the executor wrote its run log, so an immediate workflows runs wait could find no run. The call now answers running only once the run's log exists, re-reading it each second until then or until the run settles; the dispatch hook is gone. --- .../tools/handlers/workflow/mutations.test.ts | 35 +++++++++++++++---- .../tools/handlers/workflow/mutations.ts | 22 ++++++------ .../application/run-workflow-from-copilot.ts | 3 -- 3 files changed, 40 insertions(+), 20 deletions(-) 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 e2c83f9cd48..7c095189870 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts @@ -15,9 +15,14 @@ const { mocks } = vi.hoisted(() => ({ executeWorkflowUseCase: vi.fn(), hasExecutionResult: vi.fn(), readAttemptedExecutionId: vi.fn(), + logStatus: vi.fn(), }, })) +vi.mock('@/lib/workflows/executor/execution-state', () => ({ + getWorkflowExecutionLogStatus: mocks.logStatus, +})) + vi.mock('@/lib/mothership/application/execute-workflow-use-case', () => ({ executeCopilotWorkflowUseCase: mocks.executeWorkflowUseCase, messageForCopilotWorkflowError: (error: unknown, fallback = 'Workflow operation failed') => @@ -445,12 +450,9 @@ describe('a workflow tool run past its wait', () => { const started = new Promise((resolve) => { settle = resolve }) - mocks.executeWorkflowUseCase.mockImplementationOnce( - (_context: unknown, _useCase: unknown, input: { lifecycle: { onDispatch?: () => void } }) => { - if (dispatch) input.lifecycle.onDispatch?.() - return started - } - ) + mocks.executeWorkflowUseCase.mockReturnValueOnce(started) + // The run's log exists once it started; a run still in preflight has none. + mocks.logStatus.mockResolvedValue(dispatch ? 'running' : undefined) return { settle, lifecycle: () => @@ -493,7 +495,7 @@ describe('a workflow tool run past its wait', () => { await expect(held[0]).resolves.toMatchObject({ success: true }) }) - it('answers a run still in preflight past its wait with how it settles', async () => { + it('waits on a run past its wait until its log exists, answering one that never starts with how it settles', async () => { const run = pendingRun(false) let answered = false const answer = executeRunWorkflow( @@ -509,6 +511,25 @@ describe('a workflow tool run past its wait', () => { await expect(answer).resolves.toMatchObject({ success: false }) }) + it('answers a run still starting up running once its log exists', async () => { + const run = pendingRun() + mocks.logStatus.mockReset().mockResolvedValueOnce(undefined).mockResolvedValue('running') + let answered = false + const answer = executeRunWorkflow( + { workflowId: 'workflow-1' }, + { ...context, workflowRunWaitMs: 1_000 } + ).then((result) => { + answered = true + return result + }) + await vi.advanceTimersByTimeAsync(1_000) + expect(answered).toBe(false) + await vi.advanceTimersByTimeAsync(1_000) + 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 with the result when the run settles within its wait', async () => { const run = pendingRun() const answer = executeRunBlock( diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts index 0d92b8557fc..f64673eaf5b 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts @@ -39,6 +39,7 @@ import { runWorkflowFromCopilot, runWorkflowUntilBlockFromCopilot, } from '@/lib/workflows/application/run-workflow-from-copilot' +import { getWorkflowExecutionLogStatus } from '@/lib/workflows/executor/execution-state' import { hasExecutionResult, readAttemptedExecutionId } from '@/executor/utils/errors' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' @@ -257,6 +258,9 @@ 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 + /** * Runs a workflow tool's run and answers with its result. When the call names a wait the run * outlasts (`workflowRunWaitMs`), it answers the run still running by the execution id it gave @@ -272,20 +276,18 @@ async function answerWithinWait( const waitMs = context.workflowRunWaitMs if (waitMs === undefined) return run(lifecycle) const executionId = lifecycle.boundExecution?.executionId ?? generateId() - let markDispatched: () => void = () => {} - const dispatched = new Promise((resolve) => { - markDispatched = () => resolve(null) - }) - const settled = run({ ...lifecycle, executionId, onDispatch: () => markDispatched() }) + const settled = run({ ...lifecycle, executionId }) retainSandboxExecution(settled) const stopWaiting = new AbortController() - const waited = interruptibleSleep(waitMs, stopWaiting.signal).then(() => null) + const waited = (ms: number) => interruptibleSleep(ms, stopWaiting.signal).then(() => null) try { - const result = await Promise.race([settled, waited]) + let result = await Promise.race([settled, 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 getWorkflowExecutionLogStatus(executionId, workflowId))) { + result = await Promise.race([settled, waited(RUN_LOG_POLL_MS)]) + } if (result) return result - // Only a run handed to the executor is answered running; one still in preflight settles first. - const started = await Promise.race([settled, dispatched]) - if (started) return started } finally { stopWaiting.abort() } diff --git a/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts b/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts index 080cc0308ec..dcf07c96f1b 100644 --- a/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts +++ b/apps/sim/lib/workflows/application/run-workflow-from-copilot.ts @@ -65,8 +65,6 @@ export interface CopilotWorkflowRunLifecycle { * before it settles (a workflow tool that answers a still-running run by its id). */ executionId?: string - /** Called once authorization and admission passed and the run is handed to the executor. */ - onDispatch?: () => void } interface BaseCopilotRunInput { @@ -309,7 +307,6 @@ async function executeCopilotRun(params: { * can already make with the id it was handed. Keep the executor call first: anything * inserted above it would be reported as a run that may exist. */ - params.input.lifecycle.onDispatch?.() try { const result = await executeWorkflow( { From b0e72956a10c016ecf78819ea601e93adfc76f97 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:45:04 -0700 Subject: [PATCH 10/14] fix(mothership): keep waiting on a workflow tool's run when its log cannot be read --- .../tools/handlers/workflow/mutations.test.ts | 18 ++++++++++++++++++ .../tools/handlers/workflow/mutations.ts | 16 +++++++++++++++- 2 files changed, 33 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 7c095189870..e1d3593109f 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.test.ts @@ -530,6 +530,24 @@ describe('a workflow tool run past its wait', () => { run.settle({ success: true, output: {}, logs: [] }) }) + 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')) + let answered = false + const answer = executeRunWorkflow( + { workflowId: 'workflow-1' }, + { ...context, workflowRunWaitMs: 1_000 } + ).then((result) => { + answered = true + return result + }) + await vi.advanceTimersByTimeAsync(5_000) + expect(answered).toBe(false) + const executionId = run.lifecycle().executionId + run.settle({ success: true, output: { value: 3 }, logs: [], metadata: { executionId } }) + await expect(answer).resolves.toMatchObject({ success: true, output: { output: { value: 3 } } }) + }) + it('answers with the result when the run settles within its wait', async () => { const run = pendingRun() const answer = executeRunBlock( diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts index f64673eaf5b..c71665a4517 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts @@ -1,4 +1,5 @@ import { createLogger } from '@sim/logger' +import { toError } from '@sim/utils/errors' import { interruptibleSleep } from '@sim/utils/helpers' import { generateId } from '@sim/utils/id' import { filterUndefined, isPlainRecord, isRecordLike } from '@sim/utils/object' @@ -261,6 +262,19 @@ 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 +/** 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 { + return (await getWorkflowExecutionLogStatus(executionId, workflowId)) !== undefined + } catch (error) { + logger.warn('Could not read a workflow tool run past its wait; still waiting on it', { + executionId, + error: toError(error).message, + }) + return false + } +} + /** * Runs a workflow tool's run and answers with its result. When the call names a wait the run * outlasts (`workflowRunWaitMs`), it answers the run still running by the execution id it gave @@ -284,7 +298,7 @@ async function answerWithinWait( let result = await Promise.race([settled, 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 getWorkflowExecutionLogStatus(executionId, workflowId))) { + while (!result && !(await runIsReadable(executionId, workflowId))) { result = await Promise.race([settled, waited(RUN_LOG_POLL_MS)]) } if (result) return result From 8c1169fe77f98f340d3b3b14ebb34558c52d9fc6 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:52:24 -0700 Subject: [PATCH 11/14] fix(mothership): answer a browser run still running only once its log exists, and never discard a run that settled meanwhile --- .../mothership/request/tools/workflow-client-fallback.ts | 9 ++++++++- .../tools/workflow-client-settlement.integration.ts | 4 +++- .../lib/mothership/tools/handlers/workflow/mutations.ts | 2 ++ 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts index dee7b3ec7c1..5edfd24c10f 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts @@ -24,6 +24,7 @@ import { createStructuralWorkflowToolCompletionData, getWorkflowToolCompletionMessage, } from '@/lib/mothership/tools/workflow-tools' +import { getWorkflowExecutionLogStatus } from '@/lib/workflows/executor/execution-state' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' const logger = createLogger('CopilotWorkflowClientFallback') @@ -67,7 +68,8 @@ const CLIENT_CLAIM_POLL_MS = 5000 * Once `waitTimeoutMs` has passed, settles a browser-run workflow call as running in the * background, the same detach a browser sends on `pagehide`: its waiter answers the call with * the run's ids and the browser's run goes on, bound to the call, so a Stop still cancels it. - * A call no browser has claimed yet is re-read until one does or the wait ends. + * A call no browser has claimed yet, or whose run has no log yet, is re-read until it does or the + * wait ends. */ async function detachClientRunAfterWait(params: { toolCallId: string @@ -86,6 +88,11 @@ async function detachClientRunAfterWait(params: { if (!row || row.status !== ASYNC_TOOL_STATUS.running) return const executionId = getClaimedWorkflowExecutionId(row.claimedBy) if (!executionId || row.clientWorkflowExecutionId !== executionId) continue + // The claim precedes the run's log; answer running only once the run is readable by its id. + const logged = await getWorkflowExecutionLogStatus(executionId, workflowId).catch( + () => undefined + ) + if (!logged) continue const status = ASYNC_TOOL_CONFIRMATION_STATUS.background await settleClientToolCall({ toolCallId, 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 e730df540f6..20d7806c949 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 @@ -156,7 +156,7 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => async function executionLog( toolCallId: string, executionId: string, - status: 'completed' | 'failed' | 'cancelled' | 'pending' + status: 'completed' | 'failed' | 'cancelled' | 'pending' | 'running' ) { const snapshotId = generateId() snapshotIds.push(snapshotId) @@ -496,6 +496,8 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => it("settles a browser run that outlasts the worker's wait as running, bound to the call", async () => { const { toolCallId, executionId } = await claimed() + // The execute route's log of the browser's run, still going: what the worker waits on. + await executionLog(toolCallId, executionId, 'running') const runOnServer = vi.fn() const outcome = await raceWorkflowToolClientPickup({ diff --git a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts index c71665a4517..e3bc642e2cd 100644 --- a/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts +++ b/apps/sim/lib/mothership/tools/handlers/workflow/mutations.ts @@ -301,6 +301,8 @@ async function answerWithinWait( while (!result && !(await runIsReadable(executionId, workflowId))) { result = await Promise.race([settled, 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() From 41e4d9a7d83fd96d9b7a8fb35f7cd63e2edf4fac Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 01:59:52 -0700 Subject: [PATCH 12/14] fix(mothership): retry a failed call read in the browser-run detacher --- .../lib/mothership/request/tools/workflow-client-fallback.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts index 5edfd24c10f..8122a508e3a 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts @@ -84,7 +84,9 @@ async function detachClientRunAfterWait(params: { await interruptibleSleep(delayMs, signal) delayMs = CLIENT_CLAIM_POLL_MS if (signal.aborted) return - const row = await getAsyncToolCall(toolCallId) + // A failed read is retried on the next poll; only a call that left `running` ends the watch. + const row = await getAsyncToolCall(toolCallId).catch(() => undefined) + if (row === undefined) continue if (!row || row.status !== ASYNC_TOOL_STATUS.running) return const executionId = getClaimedWorkflowExecutionId(row.claimedBy) if (!executionId || row.clientWorkflowExecutionId !== executionId) continue From 20a3a28d49403f8d8e4f216d89cf3f7b8ec7b063 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 02:06:28 -0700 Subject: [PATCH 13/14] fix(mothership): keep watching a browser-run call that is still pending --- .../mothership/request/tools/workflow-client-fallback.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts index 8122a508e3a..25b319043f3 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts @@ -84,10 +84,13 @@ async function detachClientRunAfterWait(params: { await interruptibleSleep(delayMs, signal) delayMs = CLIENT_CLAIM_POLL_MS if (signal.aborted) return - // A failed read is retried on the next poll; only a call that left `running` ends the watch. + // A failed read is retried on the next poll. const row = await getAsyncToolCall(toolCallId).catch(() => undefined) if (row === undefined) continue - if (!row || row.status !== ASYNC_TOOL_STATUS.running) return + if (!row) return + // A call awaiting approval or pickup is re-read; one that settled or detached ends the watch. + if (row.status === ASYNC_TOOL_STATUS.pending) continue + if (row.status !== ASYNC_TOOL_STATUS.running) return const executionId = getClaimedWorkflowExecutionId(row.claimedBy) if (!executionId || row.clientWorkflowExecutionId !== executionId) continue // The claim precedes the run's log; answer running only once the run is readable by its id. From 08c08f04e76346ceb7a43661639e2adf37075c8b Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Sat, 31 Oct 2026 02:13:09 -0700 Subject: [PATCH 14/14] fix(mothership): detach only a browser run still going, and retry a failed detach --- .../request/tools/workflow-client-fallback.ts | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts index 25b319043f3..3503ab696d0 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts @@ -93,21 +93,29 @@ async function detachClientRunAfterWait(params: { if (row.status !== ASYNC_TOOL_STATUS.running) return const executionId = getClaimedWorkflowExecutionId(row.claimedBy) if (!executionId || row.clientWorkflowExecutionId !== executionId) continue - // The claim precedes the run's log; answer running only once the run is readable by its id. + // The claim precedes the run's log: answer running only once the run is readable by its id + // and still going. A run that already ended is reported by its own settlement. const logged = await getWorkflowExecutionLogStatus(executionId, workflowId).catch( () => undefined ) - if (!logged) continue + if (logged !== 'running') continue const status = ASYNC_TOOL_CONFIRMATION_STATUS.background - await settleClientToolCall({ + const settled = await settleClientToolCall({ toolCallId, status, message: getWorkflowToolCompletionMessage(status), data: createStructuralWorkflowToolCompletionData(status, workflowId, executionId), executionId, guard: { kind: 'open' }, + }).catch((error) => { + logger.warn('Could not settle a lapsed browser workflow wait as running; retrying', { + toolCallId, + error: toError(error).message, + }) + return undefined }) - return + // A write that failed is retried on the next poll; one another transition beat ends the watch. + if (settled !== undefined) return } }