diff --git a/ui/src/features/agent/wizardWorkflowRuntime.ts b/ui/src/features/agent/wizardWorkflowRuntime.ts index fcc920f0..ce709321 100644 --- a/ui/src/features/agent/wizardWorkflowRuntime.ts +++ b/ui/src/features/agent/wizardWorkflowRuntime.ts @@ -839,12 +839,24 @@ export class WizardWorkflowRuntime { return workflow } + private ownsOpen(openSequence: number, workspace: string): boolean { + return this.openSequence === openSequence && this.workspace === workspace + } + private async persistAndEmit(workflow: WizardWorkflowRecord): Promise { await this.persist() - this.emit(this.find(workflow.workflowId)) + 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 + // store and then poison the in-memory collection. + const targetWorkspace = this.workspace + const openSequence = this.openSequence let candidate = clone(this.collection) let lastError: unknown // Workflow updates can race with conversation/library autosaves in another @@ -852,14 +864,18 @@ export class WizardWorkflowRuntime { // failure; every retry still uses the server's current CAS revision. for (let attempt = 0; attempt < 3; attempt += 1) { try { - const saved = await this.persistence.save(this.workspace, clone(candidate)) - this.collection.revision = saved.revision + const saved = await this.persistence.save(targetWorkspace, clone(candidate)) + if (this.ownsOpen(openSequence, targetWorkspace)) { + this.collection.revision = saved.revision + } return } catch (error) { lastError = error - const remote = await this.persistence.load(this.workspace) + const remote = await this.persistence.load(targetWorkspace) candidate = mergeCollections(candidate, remote) - this.collection = candidate + if (this.ownsOpen(openSequence, targetWorkspace)) { + this.collection = candidate + } } } throw lastError instanceof Error ? lastError : new Error('Could not persist Wizard workflow.') diff --git a/ui/tests/wizardWorkflowRuntime.test.mjs b/ui/tests/wizardWorkflowRuntime.test.mjs index 46e6cd46..bde51214 100644 --- a/ui/tests/wizardWorkflowRuntime.test.mjs +++ b/ui/tests/wizardWorkflowRuntime.test.mjs @@ -16,6 +16,54 @@ function memoryPersistence() { } } +function workspaceMemoryPersistence() { + const stores = new Map() + const store = workspace => { + if (!stores.has(workspace)) stores.set(workspace, { version: 1, revision: 0, workflows: [] }) + return stores.get(workspace) + } + let holdSave = null + let conflictNextSave = false + const savedWorkspaces = [] + return { + savedWorkspaces, + async load(workspace) { return clone(store(workspace)) }, + async save(workspace, next) { + if (holdSave) { + holdSave.notifyBlocked() + await holdSave.promise + } + if (conflictNextSave) { + conflictNextSave = false + throw new Error('revision conflict') + } + const collection = store(workspace) + if (next.revision !== collection.revision) throw new Error('revision conflict') + const saved = { ...clone(next), revision: collection.revision + 1 } + stores.set(workspace, saved) + savedWorkspaces.push(workspace) + return clone(saved) + }, + snapshot(workspace) { return clone(store(workspace)) }, + conflictAndHoldNextSave() { + conflictNextSave = true + let release + let notifyBlocked + const promise = new Promise(resolve => { release = resolve }) + const blocked = new Promise(resolve => { notifyBlocked = resolve }) + holdSave = { promise, release, notifyBlocked, blocked } + }, + waitUntilSaveBlocked() { + if (!holdSave) throw new Error('conflictAndHoldNextSave() was not armed') + return holdSave.blocked + }, + releaseSaves() { + holdSave?.release() + holdSave = null + }, + } +} + function taskEvent(eventId, taskId, status, resultRefs = []) { return { event_id: eventId, @@ -283,3 +331,49 @@ test('awaiting input rejects prototype-polluting field paths before persistence' assert.match(failed.recoverableError, /requested input/) assert.equal(Object.prototype.polluted, undefined) }) + +test('an in-flight persist stays on the source workspace after open() retargets', async () => { + const { WizardWorkflowRuntime } = await import('../src/features/agent/wizardWorkflowRuntime.ts') + const persistence = workspaceMemoryPersistence() + 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' } }, + }], + }) + + await runtime.open('workspace-a') + await runtime.start({ + workflowId: 'workflow-a', type: 'source_job', workspace: 'workspace-a', + userRequest: 'Keep this checkpoint 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') + assert.equal(runtime.get('workflow-b'), undefined) + + persistence.conflictAndHoldNextSave() + const persistDuringSwitch = runtime.handleTaskEvent(taskEvent(70, 'task-source', 'running')) + await persistence.waitUntilSaveBlocked() + await runtime.open('workspace-b') + persistence.releaseSaves() + await persistDuringSwitch + + const destination = persistence.snapshot('workspace-b') + assert.deepEqual(destination.workflows.map(item => item.workflowId), ['workflow-b']) + assert.equal(destination.workflows[0].workspace, 'workspace-b') + assert.equal(runtime.get('workflow-a'), undefined) + assert.equal(runtime.get('workflow-b').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, 'running') + assert.equal(source.workflows[0].workspace, 'workspace-a') + assert.deepEqual(persistence.savedWorkspaces.slice(-1), ['workspace-a']) +})