diff --git a/apps/sim/lib/mothership/request/tools/client.ts b/apps/sim/lib/mothership/request/tools/client.ts index 9c730c16e6f..7d0c2568816 100644 --- a/apps/sim/lib/mothership/request/tools/client.ts +++ b/apps/sim/lib/mothership/request/tools/client.ts @@ -6,6 +6,7 @@ import { type AsyncTerminalCompletionSnapshot, getTerminalConfirmationStatus, isAsyncTerminalConfirmationStatus, + isTerminalAsyncStatus, } from '@/lib/mothership/async-runs/lifecycle' import { replaceTerminalAsyncToolCallResult } from '@/lib/mothership/async-runs/repository' import { MothershipStreamV1ToolOutcome } from '@/lib/mothership/generated/mothership-stream-v1' @@ -30,13 +31,20 @@ import { presentWorkflowLogsForModel, } from '@/lib/mothership/tools/workflow-output' import { + buildPausedWorkflowToolCompletion, createStructuralWorkflowToolCompletionData, getWorkflowToolCompletionExecutionId, getWorkflowToolCompletionMessage, getWorkflowToolLaunchError, + WORKFLOW_TOOL_PAUSED_STATUS, type WorkflowToolLaunchError, } from '@/lib/mothership/tools/workflow-tools' -import { getTrustedWorkflowToolExecution } from '@/lib/workflows/executor/execution-state' +import { + getHumanPauseResumeUrl, + getTrustedWorkflowToolExecution, + getWorkflowExecutionCopilotToolCallId, + getWorkflowExecutionLogStatus, +} from '@/lib/workflows/executor/execution-state' import type { ResolvedSecretTraceRegistry } from '@/executor/utils/resolved-secret-trace-registry' const logger = createLogger('CopilotClientToolWaiter') @@ -364,6 +372,42 @@ export async function waitForWorkflowToolCompletion({ return structuralWorkflowCompletion(structuralStatus, workflowId, executionId, launchError) } + if ( + completion.status === MothershipStreamV1ToolOutcome.success && + isPlainRecord(completion.data) && + completion.data.status === WORKFLOW_TOOL_PAUSED_STATUS + ) { + // Re-read rather than trust the confirmation; a run resumed since then reads as finished. + const [resumeUrl, boundToolCallId, logStatus] = await Promise.all([ + getHumanPauseResumeUrl(executionId, workflowId), + getWorkflowExecutionCopilotToolCallId(executionId, workflowId), + getWorkflowExecutionLogStatus(executionId, workflowId), + ]).catch((error) => { + logger.warn('Failed to read paused workflow tool execution', { + toolCallId, + workflowId, + executionId, + error: getErrorMessage(error), + }) + return [null, undefined, undefined] as const + }) + if (boundToolCallId === toolCallId) { + if (resumeUrl) { + toolRegistry?.markIncomplete('client-tool-completion-deferred') + return buildPausedWorkflowToolCompletion(workflowId, executionId, resumeUrl) + } + // The person answered before this read and the resumed run has no result yet. + if (logStatus !== undefined && !isTerminalAsyncStatus(logStatus)) { + toolRegistry?.markIncomplete('client-tool-completion-deferred') + return structuralWorkflowCompletion( + ASYNC_TOOL_CONFIRMATION_STATUS.background, + workflowId, + executionId + ) + } + } + } + try { trustedExecution = await getTrustedWorkflowToolExecution(executionId, workflowId, toolCallId) } catch (error) { 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 3503ab696d0..8fffd112a91 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-fallback.ts @@ -20,6 +20,7 @@ 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 { reportPausedClientWorkflowTool } from '@/lib/mothership/request/tools/workflow-client-settlement' import { createStructuralWorkflowToolCompletionData, getWorkflowToolCompletionMessage, @@ -98,6 +99,16 @@ async function detachClientRunAfterWait(params: { const logged = await getWorkflowExecutionLogStatus(executionId, workflowId).catch( () => undefined ) + // A paused run's log stays `pending`. The execute route answers a run waiting on a person; + // this covers that best-effort report failing. + if (logged === 'pending') { + const answered = await reportPausedClientWorkflowTool({ + toolCallId, + executionId, + workflowId, + }).catch(() => false) + if (answered) return + } if (logged !== 'running') continue const status = ASYNC_TOOL_CONFIRMATION_STATUS.background const settled = await settleClientToolCall({ 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 5f35f02f68b..1f675be2a67 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 @@ -42,7 +42,10 @@ import { settleClientWorkflowToolExecution, upsertAsyncToolCall, } from '@/lib/mothership/async-runs/repository' -import { getToolConfirmation } from '@/lib/mothership/persistence/tool-confirm' +import { + getToolConfirmation, + publishToolConfirmation, +} 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 { @@ -207,10 +210,13 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => } /** - * A claimed call whose bound execution paused at a human approval: the execute route settled the - * claim and its own report left the call running, as it does for every paused execution. + * A claimed call whose bound execution paused (a paused run's log stays `pending`) and whose + * execute route settled the claim, before the route reports it. */ - async function claimedAndPaused(snapshot: string = UNRESUMABLE_SNAPSHOT) { + async function claimedAndPausedUnreported( + pauseKind: 'human' | 'time', + snapshot: string = UNRESUMABLE_SNAPSHOT + ) { const { toolCallId, executionId } = await claimed() await executionLog(toolCallId, executionId, 'pending') await db.insert(pausedExecutions).values({ @@ -226,17 +232,26 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => registeredAt: new Date().toISOString(), resumeStatus: 'paused', snapshotReady: true, - pauseKind: 'human', + pauseKind, }, }, totalPauseCount: 1, }) await settleClientWorkflowToolExecution(toolCallId, executionId) - await reportSettledClientWorkflowTool({ toolCallId, executionId, workflowId }) - expect(await toolRow(toolCallId)).toMatchObject({ status: 'running', result: null }) return { toolCallId, executionId } } + /** + * A claimed call whose bound execution paused until a scheduled time: the run resumes by + * itself, so the execute route's report leaves the call held for the resume to settle. + */ + async function claimedAndPaused(snapshot: string = UNRESUMABLE_SNAPSHOT) { + const ids = await claimedAndPausedUnreported('time', snapshot) + await reportSettledClientWorkflowTool({ ...ids, workflowId }) + expect(await toolRow(ids.toolCallId)).toMatchObject({ status: 'running', result: null }) + return ids + } + async function setLogStatus(executionId: string, status: 'completed' | 'failed' | 'cancelled') { await db .update(workflowExecutionLogs) @@ -461,6 +476,142 @@ describe.runIf(Boolean(redisUrl))('settled client-claimed workflow tools', () => expect(await toolRow(toolCallId)).toMatchObject({ status: 'failed', claimedBy: null }) }) + it('answers a run paused for a person with the link to its form, and keeps that answer when the run is later cancelled', async () => { + const { toolCallId, executionId } = await claimedAndPausedUnreported('human') + const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS }) + + await reportSettledClientWorkflowTool({ toolCallId, executionId, workflowId }) + + const paused = { + success: true, + status: 'paused', + workflowId, + executionId, + resumeUrl: `http://localhost:3000/resume/${workflowId}/${executionId}`, + } + const completion = await waiting + expect(completion).toMatchObject({ status: 'success', data: paused }) + expect(completion?.message).toContain(paused.resumeUrl) + const settled = await toolRow(toolCallId) + expect(settled).toMatchObject({ status: 'completed', claimedBy: null, result: paused }) + expect(await publishedConfirmation(toolCallId)).toMatchObject({ + status: 'success', + executionId, + }) + + // The run is cancelled before anyone submits: its own report finds the call already answered. + expect(await PauseResumeManager.beginPausedCancellation(executionId, workflowId)).toBe(true) + expect(await PauseResumeManager.completePausedCancellation(executionId, workflowId)).toBe(true) + expect(await toolRow(toolCallId)).toMatchObject({ + status: 'completed', + completedAt: settled.completedAt, + result: paused, + }) + }) + + it("answers a paused run the worker's lapsed wait finds when the route's report never landed", async () => { + const { toolCallId, executionId } = await claimedAndPausedUnreported('human') + const runOnServer = vi.fn() + + const outcome = await raceWorkflowToolClientPickup({ + toolCallId, + workflowId, + timeoutMs: WAIT_MS, + graceMs: WAIT_MS, + waitTimeoutMs: 500, + runOnServer, + }) + + expect(outcome).toMatchObject({ + winner: 'client', + completion: { status: 'success', data: { status: 'paused', workflowId, executionId } }, + }) + expect(runOnServer).not.toHaveBeenCalled() + expect(await toolRow(toolCallId)).toMatchObject({ status: 'completed', claimedBy: null }) + }) + + it('answers a paused call still running when the person submitted before Chat read the pause', async () => { + const { toolCallId, executionId } = await claimedAndPausedUnreported('human') + await reportSettledClientWorkflowTool({ toolCallId, executionId, workflowId }) + // The resume took the pause point and the resumed run is going. + const [pausedRow] = await db + .select({ id: pausedExecutions.id, pausePoints: pausedExecutions.pausePoints }) + .from(pausedExecutions) + .where(eq(pausedExecutions.executionId, executionId)) + await db + .update(pausedExecutions) + .set({ + pausePoints: { + [APPROVAL_CONTEXT_ID]: { + ...(pausedRow.pausePoints as Record>)[ + APPROVAL_CONTEXT_ID + ], + resumeStatus: 'resuming', + }, + }, + }) + .where(eq(pausedExecutions.id, pausedRow.id)) + await db + .update(workflowExecutionLogs) + .set({ status: 'running' }) + .where(eq(workflowExecutionLogs.executionId, executionId)) + + const completion = await waitForWorkflowToolCompletion({ + toolCallId, + workflowId, + timeoutMs: WAIT_MS, + }) + + expect(completion).toMatchObject({ + status: 'background', + data: { status: 'running', workflowId, executionId }, + }) + }) + + it("never hands a call the link to another call's paused run", async () => { + const other = await claimedAndPausedUnreported('human') + const { toolCallId } = await claimed() + const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS }) + const claimedPause = { status: 'paused', workflowId, executionId: other.executionId } + + await completeAsyncToolCall({ toolCallId, status: 'completed', result: claimedPause }) + publishToolConfirmation({ + toolCallId, + status: 'success', + timestamp: new Date().toISOString(), + data: claimedPause, + executionId: other.executionId, + }) + + const completion = await waiting + expect(completion?.data).not.toHaveProperty('resumeUrl') + expect(JSON.stringify(completion)).not.toContain('/resume/') + }) + + it('answers from the run log when a confirmation claims a pause the run is not in', async () => { + const { toolCallId, executionId } = await claimedAndSettled() + const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS }) + const claimedPause = { + status: 'paused', + workflowId, + executionId, + resumeUrl: 'https://elsewhere.test', + } + + await completeAsyncToolCall({ toolCallId, status: 'completed', result: claimedPause }) + publishToolConfirmation({ + toolCallId, + status: 'success', + timestamp: new Date().toISOString(), + data: claimedPause, + executionId, + }) + + const completion = await waiting + expect(completion?.status).toBe('success') + expect(completion?.data).toEqual({ success: true, workflowId, executionId }) + }) + it('moves a queued async run to the background when the browser never reports', async () => { const { toolCallId, executionId } = await claimed({ workflowId, async: true }) const waiting = waitForWorkflowToolCompletion({ toolCallId, workflowId, timeoutMs: WAIT_MS }) diff --git a/apps/sim/lib/mothership/request/tools/workflow-client-settlement.ts b/apps/sim/lib/mothership/request/tools/workflow-client-settlement.ts index c43c2b7847f..d6f493f61fd 100644 --- a/apps/sim/lib/mothership/request/tools/workflow-client-settlement.ts +++ b/apps/sim/lib/mothership/request/tools/workflow-client-settlement.ts @@ -1,5 +1,7 @@ import { ASYNC_TOOL_CONFIRMATION_STATUS, + type AsyncConfirmationStatus, + type AsyncTerminalStatus, getTerminalConfirmationStatus, isTerminalAsyncStatus, } from '@/lib/mothership/async-runs/lifecycle' @@ -9,10 +11,14 @@ import { } from '@/lib/mothership/async-runs/repository' import { publishToolConfirmation } from '@/lib/mothership/persistence/tool-confirm' import { + buildPausedWorkflowToolCompletion, createStructuralWorkflowToolCompletionData, getWorkflowToolCompletionMessage, } from '@/lib/mothership/tools/workflow-tools' -import { getWorkflowExecutionLogStatus } from '@/lib/workflows/executor/execution-state' +import { + getHumanPauseResumeUrl, + getWorkflowExecutionLogStatus, +} from '@/lib/workflows/executor/execution-state' interface ReportClientWorkflowToolParams { toolCallId: string @@ -20,6 +26,35 @@ interface ReportClientWorkflowToolParams { workflowId: string } +/** + * Completes the call the execution claimed and publishes its answer. Only the first transition + * out of `running` lands; returns whether this one did. + */ +async function completeAndPublish( + { toolCallId, executionId }: ReportClientWorkflowToolParams, + rowStatus: AsyncTerminalStatus, + answer: { status: AsyncConfirmationStatus; message: string; data: Record } +): Promise { + const completed = await completeClientWorkflowToolCall( + { + toolCallId, + status: rowStatus, + result: answer.data, + error: rowStatus === 'completed' ? null : answer.message, + }, + executionId + ) + if (!completed) return false + + publishToolConfirmation({ + toolCallId, + ...answer, + timestamp: new Date().toISOString(), + executionId, + }) + return true +} + /** * Report a browser-claimed workflow tool's outcome from the execution it bound. * @@ -32,39 +67,44 @@ interface ReportClientWorkflowToolParams { * started, which is what the browser reports from its stream error. Whichever of * this and the browser's report lands first is the one kept. */ -export async function reportSettledClientWorkflowTool({ - toolCallId, - executionId, - workflowId, -}: ReportClientWorkflowToolParams): Promise { +export async function reportSettledClientWorkflowTool( + params: ReportClientWorkflowToolParams +): Promise { + const { executionId, workflowId } = params const logStatus = await getWorkflowExecutionLogStatus(executionId, workflowId) - if (logStatus !== undefined && !isTerminalAsyncStatus(logStatus)) return + if (logStatus !== undefined && !isTerminalAsyncStatus(logStatus)) { + await reportPausedClientWorkflowTool(params) + return + } const executionStatus = logStatus ?? 'failed' const status = getTerminalConfirmationStatus(executionStatus) - const message = getWorkflowToolCompletionMessage(status) - const data = createStructuralWorkflowToolCompletionData(status, workflowId, executionId) - const completed = await completeClientWorkflowToolCall( - { - toolCallId, - status: executionStatus, - result: data, - error: executionStatus === 'completed' ? null : message, - }, - executionId - ) - if (!completed) return - - publishToolConfirmation({ - toolCallId, + await completeAndPublish(params, executionStatus, { status, - message, - timestamp: new Date().toISOString(), - data, - executionId, + message: getWorkflowToolCompletionMessage(status), + data: createStructuralWorkflowToolCompletionData(status, workflowId, executionId), }) } +/** + * Settle a browser-claimed workflow tool whose run is waiting on a person with the link to + * answer, so Chat tells the user instead of waiting on an answer nobody knows is owed. A run + * paused only on time resumes itself and keeps the call held. The call's claim on this + * execution is what binds them, so no other call can be answered here. Returns whether it settled. + */ +export async function reportPausedClientWorkflowTool( + params: ReportClientWorkflowToolParams +): Promise { + const { executionId, workflowId } = params + const resumeUrl = await getHumanPauseResumeUrl(executionId, workflowId) + if (!resumeUrl) return false + return completeAndPublish( + params, + 'completed', + buildPausedWorkflowToolCompletion(workflowId, executionId, resumeUrl) + ) +} + /** * Move a browser-claimed async run to the background once the execute route has * queued it, the same transition the browser reports after the queue accepts diff --git a/apps/sim/lib/mothership/tools/workflow-tools.ts b/apps/sim/lib/mothership/tools/workflow-tools.ts index 12dcebbddc5..9b4ebd2c059 100644 --- a/apps/sim/lib/mothership/tools/workflow-tools.ts +++ b/apps/sim/lib/mothership/tools/workflow-tools.ts @@ -171,6 +171,33 @@ export function getWorkflowToolCompletionMessage(status: AsyncConfirmationStatus return 'Workflow execution failed.' } +/** Marks a workflow tool result whose run is waiting at a human pause, not finished. */ +export const WORKFLOW_TOOL_PAUSED_STATUS = 'paused' + +/** + * The answer for a workflow tool whose run paused for a person: where to answer, not an outcome. + * The model reads only `data` from a successful call, so the instruction travels there too. + */ +export function buildPausedWorkflowToolCompletion( + workflowId: string, + executionId: string, + resumeUrl: string +) { + const message = `The workflow paused at a Human in the Loop block and is waiting for a person to respond; it has not finished. Give the user this link to review and submit the form: ${resumeUrl}. The run continues by itself once they submit, so do not resume it for them. Read the run afterwards for its result.` + return { + status: ASYNC_TOOL_CONFIRMATION_STATUS.success, + message, + data: { + success: true, + status: WORKFLOW_TOOL_PAUSED_STATUS, + workflowId, + executionId, + resumeUrl, + message, + }, + } +} + export function createStructuralWorkflowToolCompletionData( status: AsyncConfirmationStatus, workflowId?: string, diff --git a/apps/sim/lib/workflows/executor/execution-state.ts b/apps/sim/lib/workflows/executor/execution-state.ts index f9391cf6717..aff65c87840 100644 --- a/apps/sim/lib/workflows/executor/execution-state.ts +++ b/apps/sim/lib/workflows/executor/execution-state.ts @@ -1,9 +1,12 @@ import { db } from '@sim/db' -import { workflowExecutionLogs } from '@sim/db/schema' +import { pausedExecutions, workflowExecutionLogs } from '@sim/db/schema' import { filterUndefined, isRecordLike } from '@sim/utils/object' import { and, desc, eq, or, sql } from 'drizzle-orm' +import { getBaseUrl } from '@/lib/core/utils/urls' import { materializeExecutionData, TRACE_STORE_REF_KEY } from '@/lib/logs/execution/trace-store' +import { buildResumeUiUrl } from '@/executor/constants' import type { SerializableExecutionState } from '@/executor/execution/types' +import type { PausePoint } from '@/executor/types' import { isResolvedSecretTraceProvenanceV1, type ResolvedSecretTraceProvenanceV1, @@ -188,6 +191,41 @@ export async function getWorkflowExecutionCopilotToolCallId( return typeof toolCallId === 'string' ? toolCallId : undefined } +/** A paused execution's pause points, stored keyed by context id (older rows: an array). */ +export function normalizePausePoints(raw: unknown): PausePoint[] { + if (!raw) return [] + if (Array.isArray(raw)) return raw as PausePoint[] + if (typeof raw === 'object') return Object.values(raw as Record) + return [] +} + +/** + * The page where a person answers the execution's open pauses, or null when nothing waits on a + * person: the run is not paused, or is paused only on time and resumes itself. + */ +export async function getHumanPauseResumeUrl( + executionId: string, + workflowId: string +): Promise { + const [paused] = await db + .select({ status: pausedExecutions.status, pausePoints: pausedExecutions.pausePoints }) + .from(pausedExecutions) + .where( + and( + eq(pausedExecutions.executionId, executionId), + eq(pausedExecutions.workflowId, workflowId) + ) + ) + .limit(1) + if (!paused || (paused.status !== 'paused' && paused.status !== 'partially_resumed')) { + return null + } + const waitsOnPerson = normalizePausePoints(paused.pausePoints).some( + (point) => point.resumeStatus === 'paused' && point.pauseKind !== 'time' + ) + return waitsOnPerson ? buildResumeUiUrl(getBaseUrl(), workflowId, executionId) : null +} + /** * The message an error finalization (`force_failed`) stored as the run's `finalOutput.error`. * Any other run's `finalOutput.error` is the workflow's own data, not a failure message. diff --git a/apps/sim/lib/workflows/executor/execution-status.ts b/apps/sim/lib/workflows/executor/execution-status.ts index 8137808fcdf..ec854781039 100644 --- a/apps/sim/lib/workflows/executor/execution-status.ts +++ b/apps/sim/lib/workflows/executor/execution-status.ts @@ -14,6 +14,7 @@ import { RESUME_EXECUTION_JOB_ID_PREFIX, WORKFLOW_EXECUTION_JOB_ID_PREFIX, } from '@/lib/workflows/executor/execution-job-ids' +import { normalizePausePoints } from '@/lib/workflows/executor/execution-state' import { getAutomaticResumeWaitingMetadata } from '@/lib/workflows/executor/paused-execution-metadata' import type { PausePoint } from '@/executor/types' @@ -69,13 +70,6 @@ function pickEarliestPausePoint(points: PausePoint[]): PausePoint | null { }, null) } -function normalizePausePoints(raw: unknown): PausePoint[] { - if (!raw) return [] - if (Array.isArray(raw)) return raw as PausePoint[] - if (typeof raw === 'object') return Object.values(raw as Record) - return [] -} - function extractError(executionData: unknown): string | null { if (!executionData || typeof executionData !== 'object') return null const data = executionData as ExecutionDataShape