From dffd9f8581f714ed75538ac206bfac1e1c397f50 Mon Sep 17 00:00:00 2001 From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> Date: Thu, 15 Oct 2026 07:13:19 +0530 Subject: [PATCH 1/2] FIX: let concurrent chat converter runs commit their own pieces The commit at the end of runConversion rejected the whole update when the state run id moved, so a conversion started in the same tick as another one lost its results even though the per input revision checks already guard staleness. Track which run owns each input, commit the outcomes a run still owns, and clear ownership wherever results are reset. The converting flag now only clears when no run is active. Fixes #3224 Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> --- frontend/src/hooks/useChatConverters.test.tsx | 53 +++++++++++++++++++ frontend/src/hooks/useChatConverters.ts | 17 +++++- 2 files changed, 68 insertions(+), 2 deletions(-) diff --git a/frontend/src/hooks/useChatConverters.test.tsx b/frontend/src/hooks/useChatConverters.test.tsx index fec26dba79..a0faccd635 100644 --- a/frontend/src/hooks/useChatConverters.test.tsx +++ b/frontend/src/hooks/useChatConverters.test.tsx @@ -170,3 +170,56 @@ describe('useChatConverters across runtime generation changes', () => { expect(result.current.stageResults['text']).toBeUndefined() }) }) + +describe('useChatConverters concurrent conversion runs', () => { + it('keeps the first piece result when a second conversion starts in the same tick', async () => { + let previewCall = 0 + mockedPreview.mockImplementation(({ original_value }: { original_value: string }) => { + previewCall += 1 + return Promise.resolve({ + original_value, + steps: [ + { + converter_id: 'base64-default', converter_type: 'Base64Converter', + input_value: original_value, input_data_type: 'text', + output_value: `out-${previewCall}`, output_data_type: 'text', + }, + ], + }) + }) + const attachments = [{ + draftId: 'att-1', type: 'image', name: 'photo.png', url: 'blob:photo', + mimeType: 'image/png', size: 10, sourceValue: 'photo source value', sourceDataType: 'image_path', + }] + const { result } = renderHook( + ({ text }: { text: string }) => useChatConverters(text, attachments), + { initialProps: { text: 'original text' } }, + ) + act(() => { + result.current.setPipeline('text', (stages) => [...stages, { id: 'stage-1', converterId: 'base64-default' }]) + result.current.setPipeline('image', (stages) => [...stages, { id: 'stage-1', converterId: 'base64-default' }]) + }) + await act(async () => { + result.current.convert('text') + }) + await waitFor(() => { + expect(result.current.stageResults['text']?.length).toBe(1) + expect(result.current.stageResults['att-1']?.length).toBe(1) + }) + const textBefore = result.current.stageResults['text']![0].value + const attBefore = result.current.stageResults['att-1']![0].value + expect(textBefore).not.toBe(attBefore) + + // Two piece scoped reruns started in one tick: each input keeps its own update. + await act(async () => { + void result.current.convert('text') + void result.current.convert('image') + }) + + await waitFor(() => { + expect(result.current.stageResults['text']![0].value).not.toBe(textBefore) + }) + expect(result.current.stageResults['att-1']![0].value).not.toBe(attBefore) + expect(result.current.isConverting).toBe(false) + }) +}) diff --git a/frontend/src/hooks/useChatConverters.ts b/frontend/src/hooks/useChatConverters.ts index 62fdeee3a9..e1089d1aa4 100644 --- a/frontend/src/hooks/useChatConverters.ts +++ b/frontend/src/hooks/useChatConverters.ts @@ -179,8 +179,13 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str })) const nextRunId = useRef(0) const activeRun = useRef(null) + // Which run currently owns each input's outcome slot. Two conversions started in + // the same tick own different inputs, so each must be allowed to commit its own + // results instead of the later run superseding the earlier one's whole commit. + const runOwnership = useRef>(new Map()) if (state.scopeKey !== scopeKey) { + runOwnership.current.clear() setState({ ...reconcileInputs(state, inputs), scopeKey, generation, stageResults: {}, errors: {}, applied: {}, workingInputs: {}, runId: state.runId + 1, isConverting: false, @@ -190,6 +195,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str // edits follow the same rule reconcileInputs applies: they survive only // when their underlying piece is unchanged, so a stale edit can never be // applied to text the user swapped in with the new generation. + runOwnership.current.clear() const next = reconcileInputs(state, inputs) setState({ ...next, @@ -291,6 +297,9 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str const runId = ++nextRunId.current activeRun.current = runId const affected = new Set(selected.map(({ input }: ConversionJob) => input.id)) + for (const { input } of selected) { + runOwnership.current.set(input.id, runId) + } setState((current: ConversionState) => ({ ...current, runId, @@ -329,11 +338,13 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str } })) if (activeRun.current === runId) activeRun.current = null + const stillConverting = activeRun.current !== null setState((current: ConversionState) => { - if (current.runId !== runId) return current const stageResults = { ...current.stageResults } const errors = { ...current.errors } for (const outcome of outcomes) { + // A newer run that selected this input owns its outcome slot now. + if (runOwnership.current.get(outcome.input.id) !== runId) continue const input = current.inputs.find((candidate: VersionedInput) => candidate.id === outcome.input.id) if ( !input || input.revision !== outcome.input.revision @@ -342,7 +353,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str if (outcome.stages) stageResults[input.id] = outcome.stages else if (outcome.error) errors[input.id] = outcome.error } - return { ...current, stageResults, errors, isConverting: false } + return { ...current, stageResults, errors, isConverting: stillConverting ? current.isConverting : false } }) } @@ -360,6 +371,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str const clearAll = useCallback((): void => { activeRun.current = null + runOwnership.current.clear() setState((current: ConversionState) => ({ ...current, runId: 0, isConverting: false, workingInputs: {}, stageResults: {}, errors: {}, applied: {}, })) @@ -380,6 +392,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str pipelines?: Record, ): void => { activeRun.current = null + runOwnership.current.clear() const restoredPipelines: Record = { ...pipelines } if (!pipelines) { for (const conversion of Object.values(conversions)) { From f013ee4c06043fac0be9e0c0574cca21ddd506dd Mon Sep 17 00:00:00 2001 From: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> Date: Thu, 15 Oct 2026 09:15:02 +0530 Subject: [PATCH 2/2] FIX: track converter runs in state and keep applying busy until all finish Run ownership and the set of active runs move into the conversion state, so a scope or generation change that React abandons mid render discards the reset together with that render instead of leaking it out of a ref. The converting flag now reflects every active run, so apply stays disabled until the last concurrent run commits, and each run removes only itself on completion. Follow-up to the review on #3225. Signed-off-by: Manohar Paturi <186662190+ManoharPaturi@users.noreply.github.com> --- frontend/src/hooks/useChatConverters.test.tsx | 204 +++++++++++++++++- frontend/src/hooks/useChatConverters.ts | 43 ++-- 2 files changed, 221 insertions(+), 26 deletions(-) diff --git a/frontend/src/hooks/useChatConverters.test.tsx b/frontend/src/hooks/useChatConverters.test.tsx index a0faccd635..2bd093d8c8 100644 --- a/frontend/src/hooks/useChatConverters.test.tsx +++ b/frontend/src/hooks/useChatConverters.test.tsx @@ -4,10 +4,11 @@ * edits must not survive a composer-text swap, and generated results must * clear. Review pass by @romanlutz on PR #2873. */ -import { act, renderHook, waitFor } from '@testing-library/react' +import { act, render, renderHook, waitFor } from '@testing-library/react' +import { Suspense, startTransition } from 'react' import { convertersApi } from '@/services/api' -import { useChatConverters } from './useChatConverters' +import { useChatConverters, usePieceConverters } from './useChatConverters' import type { MessageAttachment } from '@/types' const runtime = { generation: 'gen-1' } @@ -172,6 +173,151 @@ describe('useChatConverters across runtime generation changes', () => { }) describe('useChatConverters concurrent conversion runs', () => { + const attachments = [{ + draftId: 'att-1', type: 'image', name: 'photo.png', url: 'blob:photo', + mimeType: 'image/png', size: 10, sourceValue: 'photo source value', sourceDataType: 'image_path', + }] + + function setupTwoPieces() { + return renderHook( + ({ text }: { text: string }) => useChatConverters(text, attachments), + { initialProps: { text: 'original text' } }, + ) + } + + function deferredPreview() { + let resolve!: (value: unknown) => void + const promise = new Promise((r) => { resolve = r }) + return { + promise, + respond: (value: string) => resolve({ + original_value: 'x', + steps: [{ + converter_id: 'base64-default', converter_type: 'Base64Converter', + input_value: 'x', input_data_type: 'text', + output_value: value, output_data_type: 'text', + }], + }), + } + } + + it('keeps converting disabled until every concurrent run finishes', async () => { + const baseOutput = (original_value: string) => (original_value.includes('photo') ? 'base-photo' : 'base-text') + mockedPreview.mockImplementation(({ original_value }: { original_value: string }) => Promise.resolve({ + original_value, + steps: [1, 2].map((step) => ({ + converter_id: 'base64-default', converter_type: 'Base64Converter', + input_value: original_value, input_data_type: 'text', + output_value: step === 2 ? baseOutput(original_value) : `stage-one-of-${baseOutput(original_value)}`, + output_data_type: 'text', + })), + })) + const { result } = setupTwoPieces() + act(() => { + result.current.setPipeline('text', (stages) => [ + ...stages, { id: 'stage-1', converterId: 'base64-default' }, { id: 'stage-2', converterId: 'base64-default' }, + ]) + result.current.setPipeline('image', (stages) => [ + ...stages, { id: 'stage-1', converterId: 'base64-default' }, { id: 'stage-2', converterId: 'base64-default' }, + ]) + }) + await act(async () => { result.current.convert('text') }) + await waitFor(() => { + expect(result.current.stageResults['text']![1].value).toBe('base-text') + expect(result.current.stageResults['att-1']![1].value).toBe('base-photo') + }) + + const textRun = deferredPreview() + const imageRun = deferredPreview() + mockedPreview.mockImplementation(({ original_value }: { original_value: string }) => + original_value.includes('photo') ? imageRun.promise : textRun.promise) + + // Two piece scoped reruns: the image one finishes first. + let started: Promise | undefined + act(() => { + started = Promise.all([ + result.current.convertRemaining('text', 'stage-1'), + result.current.convertRemaining('att-1', 'stage-1'), + ]).then(() => undefined) + }) + await act(async () => { imageRun.respond('rerun-image') }) + await waitFor(() => { + expect(result.current.stageResults['att-1']![1].value).toBe('rerun-image') + }) + // The text rerun is still pending: its result is truncated to the kept prefix. + expect(result.current.stageResults['text']!.length).toBe(1) + expect(result.current.isConverting).toBe(true) + act(() => { result.current.apply() }) + expect(result.current.applied['att-1']).toBeUndefined() + + await act(async () => { + textRun.respond('rerun-text') + await started + }) + await waitFor(() => { + expect(result.current.stageResults['text']![1].value).toBe('rerun-text') + }) + expect(result.current.isConverting).toBe(false) + act(() => { result.current.apply() }) + expect(result.current.applied['text']).toBeDefined() + expect(result.current.applied['att-1']).toBeDefined() + }) + + it('keeps the first run results when the first one finishes first', async () => { + const baseOutput = (original_value: string) => (original_value.includes('photo') ? 'base-photo' : 'base-text') + mockedPreview.mockImplementation(({ original_value }: { original_value: string }) => Promise.resolve({ + original_value, + steps: [1, 2].map((step) => ({ + converter_id: 'base64-default', converter_type: 'Base64Converter', + input_value: original_value, input_data_type: 'text', + output_value: step === 2 ? baseOutput(original_value) : `stage-one-of-${baseOutput(original_value)}`, + output_data_type: 'text', + })), + })) + const { result } = setupTwoPieces() + act(() => { + result.current.setPipeline('text', (stages) => [ + ...stages, { id: 'stage-1', converterId: 'base64-default' }, { id: 'stage-2', converterId: 'base64-default' }, + ]) + result.current.setPipeline('image', (stages) => [ + ...stages, { id: 'stage-1', converterId: 'base64-default' }, { id: 'stage-2', converterId: 'base64-default' }, + ]) + }) + await act(async () => { result.current.convert('text') }) + await waitFor(() => { + expect(result.current.stageResults['text']![1].value).toBe('base-text') + expect(result.current.stageResults['att-1']![1].value).toBe('base-photo') + }) + + const textRun = deferredPreview() + const imageRun = deferredPreview() + mockedPreview.mockImplementation(({ original_value }: { original_value: string }) => + original_value.includes('photo') ? imageRun.promise : textRun.promise) + + let started: Promise | undefined + act(() => { + started = Promise.all([ + result.current.convertRemaining('text', 'stage-1'), + result.current.convertRemaining('att-1', 'stage-1'), + ]).then(() => undefined) + }) + await act(async () => { textRun.respond('rerun-text') }) + await waitFor(() => { + expect(result.current.stageResults['text']![1].value).toBe('rerun-text') + }) + expect(result.current.stageResults['att-1']!.length).toBe(1) + expect(result.current.isConverting).toBe(true) + + await act(async () => { + imageRun.respond('rerun-image') + await started + }) + await waitFor(() => { + expect(result.current.stageResults['att-1']![1].value).toBe('rerun-image') + }) + expect(result.current.isConverting).toBe(false) + }) + it('keeps the first piece result when a second conversion starts in the same tick', async () => { let previewCall = 0 mockedPreview.mockImplementation(({ original_value }: { original_value: string }) => { @@ -223,3 +369,57 @@ describe('useChatConverters concurrent conversion runs', () => { expect(result.current.isConverting).toBe(false) }) }) + +describe('usePieceConverters interrupted render', () => { + it('keeps a pending conversion committed when a scope change render is abandoned', async () => { + const inputs = [{ + id: 'text', pieceType: 'text', name: 'Text', dataType: 'text', value: 'original text', + }] + let resolveConversion!: (value: unknown) => void + mockedPreview.mockImplementation(() => new Promise((resolve) => { resolveConversion = resolve })) + let controller: ReturnType | undefined + const capture = (next: ReturnType): void => { controller = next } + const never = new Promise(() => {}) + + function Harness({ scopeKey, suspend }: { scopeKey: string; suspend: boolean }) { + const current = usePieceConverters(inputs, scopeKey) + capture(current) + if (suspend && scopeKey === 'B') throw never + return null + } + + const { rerender } = render( + }> + ) + act(() => { + controller!.setPipeline('text', (stages) => [...stages, { id: 'stage-1', converterId: 'base64-default' }]) + }) + act(() => { + void controller!.convert('text') + }) + expect(controller!.isConverting).toBe(true) + + // Switch scope inside a transition whose render suspends and never commits. + act(() => { + startTransition(() => { + rerender(}>) + }) + }) + + // The abandoned render must not have invalidated the pending conversion. + await act(async () => { + resolveConversion({ + original_value: 'original text', + steps: [{ + converter_id: 'base64-default', converter_type: 'Base64Converter', + input_value: 'original text', input_data_type: 'text', + output_value: 'committed-value', output_data_type: 'text', + }], + }) + }) + await waitFor(() => { + expect(controller!.stageResults['text']![0].value).toBe('committed-value') + }) + expect(controller!.isConverting).toBe(false) + }) +}) diff --git a/frontend/src/hooks/useChatConverters.ts b/frontend/src/hooks/useChatConverters.ts index e1089d1aa4..90bfb5fef9 100644 --- a/frontend/src/hooks/useChatConverters.ts +++ b/frontend/src/hooks/useChatConverters.ts @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import { useCallback, useMemo, useRef, useState } from 'react' import { buildAppliedConversions } from '@/utils/conversionResults' import { buildConverterInputs } from '@/components/Chat/converterTypes' @@ -36,6 +36,8 @@ interface ConversionState { applied: Record runId: number isConverting: boolean + runOwners: Record + activeRuns: number[] } interface ConversionJob { @@ -176,26 +178,22 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str applied: {}, runId: 0, isConverting: false, + runOwners: {}, + activeRuns: [], })) const nextRunId = useRef(0) - const activeRun = useRef(null) - // Which run currently owns each input's outcome slot. Two conversions started in - // the same tick own different inputs, so each must be allowed to commit its own - // results instead of the later run superseding the earlier one's whole commit. - const runOwnership = useRef>(new Map()) if (state.scopeKey !== scopeKey) { - runOwnership.current.clear() setState({ ...reconcileInputs(state, inputs), scopeKey, generation, stageResults: {}, errors: {}, applied: {}, workingInputs: {}, runId: state.runId + 1, isConverting: false, + runOwners: {}, activeRuns: [], }) } else if (state.generation !== generation) { // A runtime generation change invalidates every generated result. Working // edits follow the same rule reconcileInputs applies: they survive only // when their underlying piece is unchanged, so a stale edit can never be // applied to text the user swapped in with the new generation. - runOwnership.current.clear() const next = reconcileInputs(state, inputs) setState({ ...next, @@ -205,14 +203,13 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str applied: {}, runId: state.runId + 1, isConverting: false, + runOwners: {}, + activeRuns: [], }) } else if (state.sourceInputs !== inputs) { setState(reconcileInputs(state, inputs)) } - useEffect(() => { - activeRun.current = null - }, [generation, scopeKey]) const setPipeline = useCallback(( pieceType: string, @@ -272,7 +269,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str afterStageId, includeIncomplete = false, }: ConversionScope): Promise => { - if (activeRun.current !== null && activeRun.current === state.runId) return + if (state.activeRuns.length > 0 && state.activeRuns[state.activeRuns.length - 1] === state.runId) return const completed = completedResults(state) const selected = state.inputs.flatMap((input: VersionedInput): ConversionJob[] => { if ( @@ -295,15 +292,16 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str }) if (selected.length === 0) return const runId = ++nextRunId.current - activeRun.current = runId const affected = new Set(selected.map(({ input }: ConversionJob) => input.id)) - for (const { input } of selected) { - runOwnership.current.set(input.id, runId) - } setState((current: ConversionState) => ({ ...current, runId, isConverting: true, + activeRuns: [...current.activeRuns, runId], + runOwners: { + ...current.runOwners, + ...Object.fromEntries(selected.map(({ input }: ConversionJob) => [input.id, runId])), + }, stageResults: { ...current.stageResults, ...Object.fromEntries(selected.map(({ input, prefix }: ConversionJob) => [input.id, prefix])), @@ -337,14 +335,13 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str return { input, pipelineRevision, error: `Conversion from stage ${start + 1} failed: ${toApiError(error).detail}` } } })) - if (activeRun.current === runId) activeRun.current = null - const stillConverting = activeRun.current !== null setState((current: ConversionState) => { + const activeRuns = current.activeRuns.filter((active: number) => active !== runId) const stageResults = { ...current.stageResults } const errors = { ...current.errors } for (const outcome of outcomes) { // A newer run that selected this input owns its outcome slot now. - if (runOwnership.current.get(outcome.input.id) !== runId) continue + if (current.runOwners[outcome.input.id] !== runId) continue const input = current.inputs.find((candidate: VersionedInput) => candidate.id === outcome.input.id) if ( !input || input.revision !== outcome.input.revision @@ -353,7 +350,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str if (outcome.stages) stageResults[input.id] = outcome.stages else if (outcome.error) errors[input.id] = outcome.error } - return { ...current, stageResults, errors, isConverting: stillConverting ? current.isConverting : false } + return { ...current, stageResults, errors, activeRuns, isConverting: activeRuns.length > 0 } }) } @@ -370,10 +367,9 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str }, []) const clearAll = useCallback((): void => { - activeRun.current = null - runOwnership.current.clear() setState((current: ConversionState) => ({ ...current, runId: 0, isConverting: false, workingInputs: {}, stageResults: {}, errors: {}, applied: {}, + runOwners: {}, activeRuns: [], })) }, []) @@ -391,8 +387,6 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str conversions: Record, pipelines?: Record, ): void => { - activeRun.current = null - runOwnership.current.clear() const restoredPipelines: Record = { ...pipelines } if (!pipelines) { for (const conversion of Object.values(conversions)) { @@ -416,6 +410,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str } return { ...next, applied: { ...conversions }, workingInputs: {}, stageResults: {}, errors: {}, runId: 0, isConverting: false, + runOwners: {}, activeRuns: [], } }) }, [])