Skip to content

Commit a6f07f0

Browse files
committed
fix(api): list a paused run's points when read by a resume attempt id
A settled resume attempt has no log row of its own, so the status read answers from the parent run it continued and can report that run paused. The run detail then read pause points by the attempt id, which has no paused record, and returned none. The status read now reports which execution's state answered it, and the run detail reads the points from that execution.
1 parent 9c0e642 commit a6f07f0

3 files changed

Lines changed: 75 additions & 12 deletions

File tree

‎apps/sim/lib/workflows/application/read-workflow-run-pause.integration.ts‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import {
1313
permissionGroupMember,
1414
permissionGroupWorkspace,
1515
permissions,
16+
resumeQueue,
1617
user,
1718
workflow,
1819
workflowDeploymentVersion,
@@ -29,6 +30,7 @@ import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'
2930
import { encryptSecret } from '@/lib/core/security/encryption'
3031
import { executeWorkflowOperation } from '@/lib/workflows/application/execute-workflow'
3132
import { readWorkflowRun } from '@/lib/workflows/application/read-workflow-run'
33+
import { PauseResumeManager } from '@/lib/workflows/executor/human-in-the-loop-manager'
3234
import type { PausePoint } from '@/executor/types'
3335

3436
const organizationId = generateId()
@@ -285,4 +287,30 @@ describe('a deployed run paused at a human-in-the-loop block', () => {
285287
const run = await readRun(restricted, runId, { includeOutput: true })
286288
expect(run.pausePoints).toEqual([expect.objectContaining({ display: null, inputFields: null })])
287289
})
290+
291+
it('lists the paused parent run points when read by a settled resume attempt id', async () => {
292+
const runId = await pausedRun()
293+
const attempt = await PauseResumeManager.enqueueOrStartResume({
294+
executionId: runId,
295+
workflowId,
296+
contextId: 'approval',
297+
resumeInput: {},
298+
userId: ownerId,
299+
})
300+
/** The attempt settled without finishing the run, which still waits at its pause. */
301+
await db
302+
.update(resumeQueue)
303+
.set({ status: 'completed', completedAt: new Date() })
304+
.where(eq(resumeQueue.newExecutionId, attempt.resumeExecutionId))
305+
306+
const run = await readRun(owner, attempt.resumeExecutionId, { includeOutput: true })
307+
expect(run).toMatchObject({ status: 'paused' })
308+
expect(run.pausePoints).toEqual([
309+
expect.objectContaining({
310+
contextId: 'approval',
311+
display: REDACTED_DISPLAY,
312+
inputFields: INPUT_FIELDS,
313+
}),
314+
])
315+
})
288316
})

‎apps/sim/lib/workflows/application/read-workflow-run.ts‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ export const readWorkflowRun = defineAuthorizedWorkflowUseCase({
8585
viewerUserId: logProjectionSubjectUserId(principal),
8686
})
8787
if (!projected) throw new OrchestrationError('not_found', 'Run not found')
88-
const { status, projection } = projected
88+
const { status, projection, stateExecutionId } = projected
8989

9090
/**
9191
* A run whose `blockOutputs` the viewer's group withholds joins the same
@@ -153,7 +153,9 @@ export const readWorkflowRun = defineAuthorizedWorkflowUseCase({
153153

154154
/**
155155
* Every pause point of a paused run, so a caller can resume any of them by
156-
* `contextId`, not only the earliest one `paused` summarizes.
156+
* `contextId`, not only the earliest one `paused` summarizes. Read from the
157+
* execution whose state answered the status, so a resume attempt reporting its
158+
* parent's pause lists the parent's points.
157159
*
158160
* A point's `display` is what the block rendered from upstream outputs, so it
159161
* follows `finalOutput`'s gating: withheld unless output was asked for and the
@@ -168,7 +170,7 @@ export const readWorkflowRun = defineAuthorizedWorkflowUseCase({
168170
? (
169171
(await readPausedRunPoints({
170172
workflowId: context.workflowId,
171-
executionId: context.runId,
173+
executionId: stateExecutionId,
172174
})) ?? []
173175
).map((point) => ({
174176
...point,

‎apps/sim/lib/workflows/executor/execution-status.ts‎

Lines changed: 42 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -251,6 +251,11 @@ function projectExecutionStatus(
251251
export interface ProjectedWorkflowExecutionStatus {
252252
status: WorkflowExecutionStatusResponse
253253
projection: LogFieldProjection
254+
/**
255+
* The execution whose recorded state answered: the run itself, or the parent run a settled
256+
* resume attempt continued. Further reads of the run's state (its pause points) use this id.
257+
*/
258+
stateExecutionId: string
254259
}
255260

256261
/**
@@ -270,14 +275,18 @@ export interface ProjectedWorkflowExecutionStatus {
270275
export async function getProjectedWorkflowExecutionStatus(
271276
input: GetWorkflowExecutionStatusInput
272277
): Promise<ProjectedWorkflowExecutionStatus | null> {
273-
const status = await readWorkflowExecutionStatus(input)
274-
if (!status) return null
278+
const resolved = await readWorkflowExecutionStatus(input)
279+
if (!resolved) return null
275280
const projection = await resolveLogFieldProjection(
276281
input.viewerUserId,
277282
input.workspaceId,
278283
input.workspaceOrganizationId
279284
)
280-
return { status: projectExecutionStatus(status, projection), projection }
285+
return {
286+
status: projectExecutionStatus(resolved.status, projection),
287+
projection,
288+
stateExecutionId: resolved.stateExecutionId,
289+
}
281290
}
282291

283292
/**
@@ -330,9 +339,18 @@ async function readRunDeploymentVersion(
330339
return row?.version ?? null
331340
}
332341

342+
/**
343+
* A status resource and the execution whose recorded state answered it: the run itself, or
344+
* for a settled resume attempt with no log row of its own, the parent run it continued.
345+
*/
346+
interface ResolvedExecutionStatus {
347+
status: WorkflowExecutionStatusResponse
348+
stateExecutionId: string
349+
}
350+
333351
async function readWorkflowExecutionStatus(
334352
input: GetWorkflowExecutionStatusInput
335-
): Promise<WorkflowExecutionStatusResponse | null> {
353+
): Promise<ResolvedExecutionStatus | null> {
336354
const { workflowId, executionId, includeOutput, selectedOutputs } = input
337355

338356
const [logRow] = await db
@@ -412,13 +430,16 @@ async function readWorkflowExecutionStatus(
412430
for (const jobId of queueJobIds) {
413431
const job = await jobQueue.getJob(jobId)
414432
if (!job || job.metadata.workflowId !== workflowId) continue
415-
return projectQueueJob(job, { executionId, includeOutput, workflowId }, deploymentVersion)
433+
return {
434+
status: projectQueueJob(job, { executionId, includeOutput, workflowId }, deploymentVersion),
435+
stateExecutionId: executionId,
436+
}
416437
}
417438
}
418439

419440
if (projectedResume) {
420441
const startedAt = projectedResume.claimedAt ?? projectedResume.queuedAt
421-
return {
442+
const status: WorkflowExecutionStatusResponse = {
422443
executionId,
423444
workflowId,
424445
status: 'queued',
@@ -434,6 +455,7 @@ async function readWorkflowExecutionStatus(
434455
finalOutput: null,
435456
blockOutputs: null,
436457
}
458+
return { status, stateExecutionId: executionId }
437459
}
438460

439461
if (!logRow) {
@@ -442,7 +464,12 @@ async function readWorkflowExecutionStatus(
442464
...input,
443465
executionId: resumeAttempt.parentExecutionId,
444466
})
445-
return run ? projectSettledResumeAttempt(executionId, resumeAttempt, run) : null
467+
return run
468+
? {
469+
status: projectSettledResumeAttempt(executionId, resumeAttempt, run.status),
470+
stateExecutionId: run.stateExecutionId,
471+
}
472+
: null
446473
}
447474

448475
const [pausedRow] = await db
@@ -509,7 +536,10 @@ async function readWorkflowExecutionStatus(
509536
const readsExecutionData =
510537
selectedOutputs.length > 0 || statusesReadingExecutionData(includeOutput).includes(status)
511538
if (!readsExecutionData) {
512-
return { ...resource, error: null, finalOutput: null, blockOutputs: null }
539+
return {
540+
status: { ...resource, error: null, finalOutput: null, blockOutputs: null },
541+
stateExecutionId: executionId,
542+
}
513543
}
514544

515545
const requestedBlockIds = [
@@ -538,7 +568,10 @@ async function readWorkflowExecutionStatus(
538568
? pickSelectedOutputs(selectedOutputs, materialized.blockOutputs)
539569
: null
540570

541-
return { ...resource, error, finalOutput, blockOutputs }
571+
return {
572+
status: { ...resource, error, finalOutput, blockOutputs },
573+
stateExecutionId: executionId,
574+
}
542575
}
543576

544577
/** Most pause points a run detail lists; a run over it lists its earliest registered points. */

0 commit comments

Comments
 (0)