FIX: let concurrent chat converter runs commit their own pieces

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' }
@@ -170,3 +171,255 @@ describe('useChatConverters across runtime generation changes', () => {
     expect(result.current.stageResults['text']).toBeUndefined()
   })
 })
+
+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<void> | 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<void> | 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 }) => {
+      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)
+  })
+})
+
+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<typeof usePieceConverters> | undefined
+    const capture = (next: ReturnType<typeof usePieceConverters>): 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(
+      <Suspense fallback={<div />}><Harness scopeKey="A" suspend={false} /></Suspense>
+    )
+    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(<Suspense fallback={<div />}><Harness scopeKey="B" suspend /></Suspense>)
+      })
+    })
+
+    // 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)
+  })
+})

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<string, PieceConversion>
   runId: number
   isConverting: boolean
+  runOwners: Record<string, number>
+  activeRuns: number[]
 }
 
 interface ConversionJob {
@@ -176,14 +178,16 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str
     applied: {},
     runId: 0,
     isConverting: false,
+    runOwners: {},
+    activeRuns: [],
   }))
   const nextRunId = useRef(0)
-  const activeRun = useRef<number | null>(null)
 
   if (state.scopeKey !== scopeKey) {
     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
@@ -199,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,
@@ -266,7 +269,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str
     afterStageId,
     includeIncomplete = false,
   }: ConversionScope): Promise<void> => {
-    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 (
@@ -289,12 +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))
     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])),
@@ -328,12 +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
     setState((current: ConversionState) => {
-      if (current.runId !== runId) return current
+      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 (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
@@ -342,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: false }
+      return { ...current, stageResults, errors, activeRuns, isConverting: activeRuns.length > 0 }
     })
   }
 
@@ -359,9 +367,9 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str
   }, [])
 
   const clearAll = useCallback((): void => {
-    activeRun.current = null
     setState((current: ConversionState) => ({
       ...current, runId: 0, isConverting: false, workingInputs: {}, stageResults: {}, errors: {}, applied: {},
+      runOwners: {}, activeRuns: [],
     }))
   }, [])
 
@@ -379,7 +387,6 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str
     conversions: Record<string, PieceConversion>,
     pipelines?: Record<string, ConverterPipelineStage[]>,
   ): void => {
-    activeRun.current = null
     const restoredPipelines: Record<string, ConverterPipelineStage[]> = { ...pipelines }
     if (!pipelines) {
       for (const conversion of Object.values(conversions)) {
@@ -403,6 +410,7 @@ export function usePieceConverters(inputs: ConverterInputPiece[], scopeKey?: str
       }
       return {
         ...next, applied: { ...conversions }, workingInputs: {}, stageResults: {}, errors: {}, runId: 0, isConverting: false,
+        runOwners: {}, activeRuns: [],
       }
     })
   }, [])