From c18ef4de6e6562c83570a602732bb48add3f5dd9 Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 8 Sep 2026 11:21:07 +0000 Subject: [PATCH] fix(wizard): persist step completion to the source workspace after open() Pin the collection that still holds the running workflow before execute() yields. Switching folders mid-step no longer drops the finished checkpoint on the source workspace or writes it into the destination. Co-authored-by: ignaciodelcano+dcl --- .../features/agent/wizardWorkflowRuntime.ts | 60 +++++++++++++------ ui/tests/wizardWorkflowRuntime.test.mjs | 57 ++++++++++++++++++ 2 files changed, 98 insertions(+), 19 deletions(-) diff --git a/ui/src/features/agent/wizardWorkflowRuntime.ts b/ui/src/features/agent/wizardWorkflowRuntime.ts index ce709321..63f17967 100644 --- a/ui/src/features/agent/wizardWorkflowRuntime.ts +++ b/ui/src/features/agent/wizardWorkflowRuntime.ts @@ -709,6 +709,10 @@ export class WizardWorkflowRuntime { const workflow = this.find(workflowId) const definition = this.definitions.get(workflow.type) if (!definition) return + // `execute()` can yield. `open()` may replace `this.collection` before the + // step writes its checkpoint, so persist must keep the collection that + // still holds this workflow object. + const owner = this.capturePersistOwner() while (workflow.currentStep < workflow.steps.length) { if (workflow.cancelRequested) return const step = workflow.steps[workflow.currentStep] @@ -717,7 +721,7 @@ export class WizardWorkflowRuntime { workflow.state = failureState(workflow) workflow.recoverableError = `Missing workflow step definition ${step.stepId}.` workflow.updatedAt = Date.now() - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) return } if (step.state === 'completed') { @@ -731,7 +735,7 @@ export class WizardWorkflowRuntime { step.state = 'failed' step.error = workflow.recoverableError workflow.updatedAt = Date.now() - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) return } const now = Date.now() @@ -741,7 +745,7 @@ export class WizardWorkflowRuntime { step.error = '' workflow.state = workflow.resumeRequested ? 'retrying' : 'running' workflow.updatedAt = now - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) let result: WizardWorkflowStepResult try { result = await stepDefinition.execute({ @@ -753,7 +757,7 @@ export class WizardWorkflowRuntime { workflow.state = failureState(workflow) workflow.recoverableError = step.error workflow.updatedAt = Date.now() - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) return } step.output = clone(result.output || {}) @@ -771,7 +775,7 @@ export class WizardWorkflowRuntime { step.completedAt = workflow.updatedAt workflow.currentStep += 1 workflow.state = workflow.currentStep >= workflow.steps.length ? 'completed' : 'running' - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) continue } if (result.state === 'awaiting_input') { @@ -781,7 +785,7 @@ export class WizardWorkflowRuntime { step.error = `Step ${step.stepId} requested input without a reason and declared fields.` workflow.state = failureState(workflow) workflow.recoverableError = step.error - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) return } const previousPending = workflow.pendingInput @@ -812,7 +816,7 @@ export class WizardWorkflowRuntime { } step.state = 'awaiting_input' workflow.state = 'awaiting_input' - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) return } if (!step.taskId) { @@ -820,17 +824,17 @@ export class WizardWorkflowRuntime { step.error = `Step ${step.stepId} returned ${result.state} without a canonical task id.` workflow.state = failureState(workflow) workflow.recoverableError = step.error - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) return } step.state = 'waiting' workflow.state = result.state === 'queued' ? 'queued' : 'waiting' - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) return } workflow.state = 'completed' workflow.updatedAt = Date.now() - await this.persistAndEmit(workflow) + await this.persistAndEmit(workflow, owner) } private find(workflowId: string): WizardWorkflowRecord { @@ -843,21 +847,37 @@ export class WizardWorkflowRuntime { return this.openSequence === openSequence && this.workspace === workspace } - private async persistAndEmit(workflow: WizardWorkflowRecord): Promise { - await this.persist() + private capturePersistOwner(): { + workspace: string + collection: WizardWorkflowCollection + openSequence: number + } { + return { + workspace: this.workspace, + collection: this.collection, + openSequence: this.openSequence, + } + } + + private async persistAndEmit( + workflow: WizardWorkflowRecord, + owner = this.capturePersistOwner(), + ): Promise { + await this.persist(owner) const current = this.collection.workflows.find(item => item.workflowId === workflow.workflowId) if (!current || current.workspace !== this.workspace) return this.emit(current) } - private async persist(): Promise { - // Pin the workspace that owns this snapshot. `open()` rebinds - // `this.workspace` immediately, so a conflict retry must not follow the - // live pointer or it will merge the source checkpoint into the destination + private async persist(owner = this.capturePersistOwner()): Promise { + // Pin the workspace and collection that own this snapshot. `open()` + // rebinds both pointers immediately. A conflict retry — or a persist + // that starts after `execute()` yields — must not follow the live + // pointer or it will write the source checkpoint into the destination // store and then poison the in-memory collection. - const targetWorkspace = this.workspace - const openSequence = this.openSequence - let candidate = clone(this.collection) + const targetWorkspace = owner.workspace + const openSequence = owner.openSequence + let candidate = clone(owner.collection) let lastError: unknown // Workflow updates can race with conversation/library autosaves in another // tab. Re-read and merge a few times before surfacing a real persistence @@ -865,6 +885,7 @@ export class WizardWorkflowRuntime { for (let attempt = 0; attempt < 3; attempt += 1) { try { const saved = await this.persistence.save(targetWorkspace, clone(candidate)) + owner.collection.revision = saved.revision if (this.ownsOpen(openSequence, targetWorkspace)) { this.collection.revision = saved.revision } @@ -873,6 +894,7 @@ export class WizardWorkflowRuntime { lastError = error const remote = await this.persistence.load(targetWorkspace) candidate = mergeCollections(candidate, remote) + owner.collection.revision = candidate.revision if (this.ownsOpen(openSequence, targetWorkspace)) { this.collection = candidate } diff --git a/ui/tests/wizardWorkflowRuntime.test.mjs b/ui/tests/wizardWorkflowRuntime.test.mjs index bde51214..46268b3b 100644 --- a/ui/tests/wizardWorkflowRuntime.test.mjs +++ b/ui/tests/wizardWorkflowRuntime.test.mjs @@ -377,3 +377,60 @@ test('an in-flight persist stays on the source workspace after open() retargets' assert.equal(source.workflows[0].workspace, 'workspace-a') assert.deepEqual(persistence.savedWorkspaces.slice(-1), ['workspace-a']) }) + +test('a step that finishes after open() still persists to the source workspace', async () => { + const { WizardWorkflowRuntime } = await import('../src/features/agent/wizardWorkflowRuntime.ts') + const persistence = workspaceMemoryPersistence() + let releaseFinish + let notifyBlocked + const finishBlocked = new Promise(resolve => { notifyBlocked = resolve }) + const finishGate = new Promise(resolve => { releaseFinish = resolve }) + + const runtime = new WizardWorkflowRuntime(persistence) + runtime.register({ + type: 'source_job', + steps: [{ + stepId: 'wait', kind: 'wait for task', + async execute() { return { state: 'waiting', taskId: 'task-source' } }, + }, { + stepId: 'finish', kind: 'publish result', + async execute() { + notifyBlocked() + await finishGate + return { state: 'completed', output: { published: true }, outputRefs: ['final.mp4'] } + }, + }], + }) + + await runtime.open('workspace-a') + await runtime.start({ + workflowId: 'workflow-a', type: 'source_job', workspace: 'workspace-a', + userRequest: 'Keep the finished step on A', + }) + await runtime.open('workspace-b') + await runtime.start({ + workflowId: 'workflow-b', type: 'source_job', workspace: 'workspace-b', + userRequest: 'Destination already has its own checkpoint', + }) + await runtime.open('workspace-a') + assert.equal(runtime.get('workflow-a').state, 'waiting') + + const completion = runtime.handleTaskEvent(taskEvent(71, 'task-source', 'completed', ['song.wav'])) + await finishBlocked + await runtime.open('workspace-b') + releaseFinish() + await completion + + const destination = persistence.snapshot('workspace-b') + assert.deepEqual(destination.workflows.map(item => item.workflowId), ['workflow-b']) + assert.equal(destination.workflows[0].state, 'waiting') + assert.equal(destination.workflows[0].workspace, 'workspace-b') + + const source = persistence.snapshot('workspace-a') + assert.deepEqual(source.workflows.map(item => item.workflowId), ['workflow-a']) + assert.equal(source.workflows[0].state, 'completed') + assert.equal(source.workflows[0].workspace, 'workspace-a') + assert.deepEqual(source.workflows[0].outputRefs, ['song.wav', 'final.mp4']) + assert.equal(runtime.get('workflow-a'), undefined) + assert.equal(runtime.get('workflow-b').workspace, 'workspace-b') +})