fix(core): audio block job

packages/backend/server/src/core/realtime/__tests__/registry.spec.ts

@@ -1065,7 +1065,6 @@ test('copilot transcript realtime provider registers task live query handlers',
       return { id: taskId, status: 'running', userId, workspaceId };
     },
   } as unknown as CopilotTranscriptionRetryService;
-
   new CopilotTranscriptRealtimeProvider(ac, transcript, retry, registry, {
     copilot: { enabled: true },
   } as never).onModuleInit();

packages/frontend/core/src/blocksuite/attachment-viewer/audio/audio-block.tsx

@@ -115,9 +115,10 @@ const AttachmentAudioPlayer = ({ block }: { block: AudioAttachmentBlock }) => {
     }
 
     setPreflightChecking(true);
-    const result = await block.transcriptionJob.preflightCheck();
-    setPreflightChecking(false);
-    if (result?.error === 'created-by-others') {
+    const result = await block.transcribe().finally(() => {
+      setPreflightChecking(false);
+    });
+    if (result?.status === 'blocked') {
       confirmModal.openConfirmModal({
         title: t['com.affine.audio.transcribe.non-owner.confirm.title'](),
         description: (
@@ -137,7 +138,6 @@ const AttachmentAudioPlayer = ({ block }: { block: AudioAttachmentBlock }) => {
         method: 'not owner',
       });
     } else {
-      await block.transcribe();
       track.doc.editor.audioBlock.transcribeRecording({
         type: 'Meeting record',
         method: 'success',

packages/frontend/core/src/modules/media/entities/audio-attachment-block.ts

@@ -101,7 +101,7 @@ export class AudioAttachmentBlock extends Entity<AttachmentBlockModel> {
       this.transcriptionJob.status$.value.status === 'waiting-for-job' &&
       !this.hasTranscription$.value
     ) {
-      this.transcribe().catch(error => {
+      this.resumeTranscription().catch(error => {
         logger.error('Error transcribing audio:', error);
       });
     }
@@ -167,16 +167,20 @@ export class AudioAttachmentBlock extends Entity<AttachmentBlockModel> {
     return job;
   }
 
-  readonly transcribe = async () => {
+  private readonly runTranscription = async (retryFailed: boolean) => {
     try {
       const initialStatus = this.transcriptionJob.status$.value.status;
       if (initialStatus !== 'waiting-for-job' && initialStatus !== 'failed') {
         return;
       }
-      const status = await this.transcriptionJob.start();
+      const status = await this.transcriptionJob.start(retryFailed);
+      if (status.status === 'blocked') {
+        return status;
+      }
       if (status.status === 'settled') {
         await this.fillTranscriptionResult(status.result);
       }
+      return status;
     } catch (error) {
       track.doc.editor.audioBlock.transcribeRecording({
         type: 'Meeting record',
@@ -187,6 +191,10 @@ export class AudioAttachmentBlock extends Entity<AttachmentBlockModel> {
     }
   };
 
+  readonly resumeTranscription = () => this.runTranscription(false);
+
+  readonly transcribe = () => this.runTranscription(true);
+
   private readonly fillTranscriptionResult = async (
     result: TranscriptionResult
   ) => {

packages/frontend/core/src/modules/media/entities/audio-transcription-job.spec.ts

@@ -0,0 +1,111 @@
+/** @vitest-environment happy-dom */
+
+import type { TranscriptionBlockProps } from '@affine/core/blocksuite/ai/blocks/transcription-block/model';
+import { AiJobStatus } from '@affine/graphql';
+import { Framework } from '@toeverything/infra';
+import { describe, expect, test, vi } from 'vitest';
+
+import { AuthService } from '../../cloud/services/auth';
+import { DefaultServerService } from '../../cloud/services/default-server';
+import { WorkspaceServerService } from '../../cloud/services/workspace-server';
+import { NbstoreService } from '../../storage';
+import { WorkspaceService } from '../../workspace';
+import { AudioTranscriptionJob } from './audio-transcription-job';
+import { AudioTranscriptionJobStore } from './audio-transcription-job-store';
+
+describe('AudioTranscriptionJob', () => {
+  test('only retries a failed task after explicit user intent', async () => {
+    const request = vi
+      .fn()
+      .mockResolvedValueOnce({
+        task: { id: 'task-1', status: AiJobStatus.failed },
+      })
+      .mockResolvedValueOnce({
+        task: { id: 'task-1', status: AiJobStatus.failed },
+      })
+      .mockResolvedValueOnce({
+        task: { id: 'task-1', status: AiJobStatus.failed },
+      });
+    const server = {
+      scope: {
+        get: () => null,
+        getOptional: (key: unknown) =>
+          key === AuthService
+            ? { session: { account$: { value: { id: 'user-1' } } } }
+            : null,
+      },
+    };
+    const framework = new Framework();
+    framework
+      .service(WorkspaceService, {
+        workspace: { id: 'workspace-1' },
+      } as WorkspaceService)
+      .service(WorkspaceServerService, {
+        server,
+      } as unknown as WorkspaceServerService)
+      .service(DefaultServerService, {
+        server: null,
+      } as unknown as DefaultServerService)
+      .service(NbstoreService, {
+        realtime: { request },
+      } as unknown as NbstoreService)
+      .entity(AudioTranscriptionJobStore, [
+        WorkspaceService,
+        WorkspaceServerService,
+        DefaultServerService,
+        NbstoreService,
+      ])
+      .entity(AudioTranscriptionJob, [
+        WorkspaceServerService,
+        DefaultServerService,
+      ]);
+
+    const job = framework.provider().createEntity(AudioTranscriptionJob, {
+      blobId: 'blob-1',
+      blockProps: {
+        jobId: 'task-1',
+        createdBy: 'user-1',
+      } as TranscriptionBlockProps,
+      getAudioTranscriptionInput: async () => ({ files: [] }),
+    });
+
+    const resumed = await job.start(false);
+
+    expect(resumed.status).toBe(AiJobStatus.failed);
+    expect(request).toHaveBeenCalledTimes(1);
+
+    const [first, second] = await Promise.all([
+      job.start(true),
+      job.start(true),
+    ]);
+
+    expect(first.status).toBe(AiJobStatus.failed);
+    expect(second).toBe(first);
+    expect(request).toHaveBeenCalledTimes(3);
+    expect(request).toHaveBeenNthCalledWith(
+      1,
+      'copilot.transcript.task.get',
+      {
+        workspaceId: 'workspace-1',
+        taskId: 'task-1',
+        blobId: 'blob-1',
+      },
+      { timeoutMs: 10000 }
+    );
+    expect(request).toHaveBeenNthCalledWith(
+      2,
+      'copilot.transcript.task.get',
+      {
+        workspaceId: 'workspace-1',
+        taskId: 'task-1',
+        blobId: 'blob-1',
+      },
+      { timeoutMs: 10000 }
+    );
+    expect(request).toHaveBeenNthCalledWith(
+      3,
+      'copilot.transcript.task.retry',
+      { workspaceId: 'workspace-1', taskId: 'task-1' }
+    );
+  });
+});

packages/frontend/core/src/modules/media/entities/audio-transcription-job.ts

@@ -26,6 +26,14 @@ export type TranscriptionStatus =
   | { status: AiJobStatus.finished }
   | { status: 'settled'; result: TranscriptionResult };
 
+export type TranscriptionStartResult =
+  | TranscriptionStatus
+  | {
+      status: 'blocked';
+      error: 'created-by-others';
+      userId: string;
+    };
+
 const logger = new DebugLogger('audio-transcription-job');
 
 function hasSettledTranscriptResult(
@@ -71,6 +79,8 @@ export class AudioTranscriptionJob extends Entity<{
     RealtimeTopicEventOf<'copilot.transcript.task.changed'>
   >;
   private taskWaitReject?: (error: unknown) => void;
+  private startPromise?: Promise<TranscriptionStartResult>;
+  private retryFailedRequested = false;
 
   private readonly _status$ = new LiveData<TranscriptionStatus>({
     status: 'waiting-for-job',
@@ -101,46 +111,36 @@ export class AudioTranscriptionJob extends Entity<{
     return null;
   });
 
-  // check if we can kick start the transcription job
-  readonly preflightCheck = async () => {
-    // if the job id is given, check if the job exists
-    if (this.props.blockProps.jobId) {
-      const existingJob = await this.store.getTranscriptTask(
-        this.props.blobId,
-        this.props.blockProps.jobId
-      );
-
-      if (hasSettledTranscriptResult(existingJob)) {
-        // if job exists, anyone can query it
-        return;
-      }
-
-      if (
-        !existingJob &&
-        this.props.blockProps.createdBy &&
-        this.props.blockProps.createdBy !== this.currentUserId
-      ) {
-        return {
-          error: 'created-by-others',
-          userId: this.props.blockProps.createdBy,
-        };
-      }
+  start(retryFailed: boolean): Promise<TranscriptionStartResult> {
+    this.retryFailedRequested ||= retryFailed;
+    if (this.startPromise) {
+      return this.startPromise;
     }
+    const promise = this.runStart();
+    this.startPromise = promise;
+    void promise.then(
+      () => {
+        if (this.startPromise === promise) {
+          this.startPromise = undefined;
+          this.retryFailedRequested = false;
+        }
+      },
+      () => {
+        if (this.startPromise === promise) {
+          this.startPromise = undefined;
+          this.retryFailedRequested = false;
+        }
+      }
+    );
+    return promise;
+  }
 
-    // if no job id, anyone can start a new job
-    return;
-  };
-
-  async start() {
+  private async runStart(): Promise<TranscriptionStartResult> {
     if (this.disposed) {
       logger.debug('Job already disposed, cannot start');
       throw new Error('Job already disposed');
     }
 
-    this._status$.value = {
-      status: 'started',
-    };
-
     try {
       // firstly check if there is a job already
       logger.debug('Checking for existing transcription job', {
@@ -150,15 +150,38 @@ export class AudioTranscriptionJob extends Entity<{
       let job: {
         id: string;
         status: AiJobStatus;
+        normalizedTranscript?: string | null;
+        transcription?: unknown[] | null;
       } | null = await this.store.getTranscriptTask(
         this.props.blobId,
         this.props.blockProps.jobId
       );
 
+      if (
+        this.props.blockProps.jobId &&
+        !hasSettledTranscriptResult(job) &&
+        !job &&
+        this.props.blockProps.createdBy &&
+        this.props.blockProps.createdBy !== this.currentUserId
+      ) {
+        return {
+          status: 'blocked',
+          error: 'created-by-others',
+          userId: this.props.blockProps.createdBy,
+        };
+      }
+
+      this._status$.value = {
+        status: 'started',
+      };
+
       if (!job) {
         logger.debug('No existing job found, submitting new transcription job');
         job = await this.store.submitTranscriptTask();
       } else if (job.status === AiJobStatus.failed) {
+        if (!this.retryFailedRequested) {
+          throw UserFriendlyError.fromAny('Transcription job failed');
+        }
         logger.debug('Found existing failed job, retrying', {
           jobId: job.id,
         });