From 81fb82761c5eb61cccbb21eacb5e4f6f1caf6916 Mon Sep 17 00:00:00 2001 From: DarkSky Date: Mon, 17 Aug 2026 01:25:09 +0800 Subject: [PATCH] feat(server): improve stability for long records --- Cargo.lock | 6 +- .../copilot/copilot-provider.spec.ts | 6 +- .../copilot/transcript-contract.spec.ts | 208 +++++++---- .../copilot/runtime/action-runtime-bridge.ts | 18 +- .../plugins/copilot/transcript/constants.ts | 3 +- .../plugins/copilot/transcript/projection.ts | 60 ++++ .../src/plugins/copilot/transcript/schema.ts | 29 +- .../src/plugins/copilot/transcript/service.ts | 331 ++++++++++++++---- 8 files changed, 506 insertions(+), 155 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 530ee2bfe9..8a306342ff 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2963,7 +2963,7 @@ dependencies = [ "libc", "log", "rustversion", - "windows-link 0.1.3", + "windows-link 0.2.1", "windows-result 0.4.1", ] @@ -4756,9 +4756,9 @@ dependencies = [ [[package]] name = "llm_adapter" -version = "0.2.19" +version = "0.2.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c17ef06aac72e31a4fae7937998029267c4d923a0671bf1a4062c96a1ab5fa70" +checksum = "5d13be366ea35d2a9966ad5770e3d070e495af4e1e6dd009638dee4b55ab45ed" dependencies = [ "base64", "jsonschema", diff --git a/packages/backend/server/src/__tests__/copilot/copilot-provider.spec.ts b/packages/backend/server/src/__tests__/copilot/copilot-provider.spec.ts index 1e2df56e2e..12fdac54f0 100644 --- a/packages/backend/server/src/__tests__/copilot/copilot-provider.spec.ts +++ b/packages/backend/server/src/__tests__/copilot/copilot-provider.spec.ts @@ -196,11 +196,7 @@ providerTest( 'managed transcript route executes the provider-neutral job port', async t => { const { models, runtime, transcript } = t.context; - await assertManagedRoute( - runtime, - 'transcript.audio', - 'Transcript audio structured' - ); + await assertManagedRoute(runtime, 'transcript.audio', 'Transcript audio'); const user = await models.user.create({ email: `copilot-provider-transcript-${randomUUID()}@affine.pro`, }); diff --git a/packages/backend/server/src/__tests__/copilot/transcript-contract.spec.ts b/packages/backend/server/src/__tests__/copilot/transcript-contract.spec.ts index 5b53c895b4..19bf6c0aed 100644 --- a/packages/backend/server/src/__tests__/copilot/transcript-contract.spec.ts +++ b/packages/backend/server/src/__tests__/copilot/transcript-contract.spec.ts @@ -63,7 +63,7 @@ test('TranscriptPayloadSchema rejects empty payloads', t => { function createTranscriptPromptService() { return { - get: Sinon.stub().resolves({ name: 'Transcript audio structured' }), + get: Sinon.stub().callsFake(async name => ({ name })), finish: Sinon.stub().callsFake((_prompt, params) => [ { role: 'user', @@ -73,57 +73,16 @@ function createTranscriptPromptService() { }; } -async function buildNativeTranscriptResult(input: any, runId: string) { - await input.onRunCreated?.({ runId, attempt: 1 }); - const nativeInput = { input: input.inputSnapshot }; - return { - nativeInput, - result: { - sourceAudio: nativeInput.input.sourceAudio ?? null, - quality: nativeInput.input.quality ?? null, - infos: [{ url: 'about:invalid', mimeType: 'text/plain', index: 0 }], - sliceManifest: null, - normalizedSegments: [ - { - speaker: 'A', - startSec: 5, - endSec: 9, - start: '00:00:05', - end: '00:00:09', - text: 'Kickoff', - }, - ], - normalizedTranscript: '00:00:05 A: Kickoff', - summaryJson: { - title: 'Weekly Sync', - durationMinutes: 1, - attendees: ['A'], - keyPoints: ['Kickoff'], - actionItems: [], - decisions: [], - openQuestions: [], - blockers: [], - }, - version: 'transcript-result-v1', - }, - }; -} - function createSuccessfulTranscriptBridge( runId: string, bridgeInputs: unknown[] ) { return { - runStream: (input: unknown) => + runStream: (input: any, executor: (input: any) => Promise) => (async function* () { - const { nativeInput, result } = await buildNativeTranscriptResult( - input, - runId - ); - bridgeInputs.push({ - ...(input as Record), - nativeInput, - }); + await input.onRunCreated?.({ runId, attempt: 1 }); + const { result } = await executor(input); + bridgeInputs.push(input); yield { type: 'action_done' as const, actionId: 'transcript.audio', @@ -334,7 +293,7 @@ test('retryTask reuses failed task and queues a new action attempt', async t => user: 'user-1', workspace: 'workspace-1', featureKind: 'transcript', - builtInRouteId: 'Transcript audio structured', + builtInRouteId: 'Transcript audio', } ); t.is(assertRoute.callCount, 2); @@ -468,13 +427,13 @@ for (const status of ['ready', 'settled']) { user: 'user-1', workspace: 'workspace-1', featureKind: 'transcript', - builtInRouteId: 'Transcript audio structured', + builtInRouteId: 'Transcript audio', } ); }); } -test('transcriptTask runs native transcript recipe through action bridge when available', async t => { +test('transcriptTask transcribes each audio slice and merges absolute timestamps', async t => { const payload = TranscriptPayloadSchema.parse({ sourceAudio: { blobId: 'blob-1', mimeType: 'audio/opus' }, sliceManifest: [ @@ -485,6 +444,13 @@ test('transcriptTask runs native transcript recipe through action bridge when av startSec: 12, durationSec: 30, }, + { + index: 1, + fileName: 'audio-1.opus', + mimeType: 'audio/opus', + startSec: 42, + durationSec: 300, + }, ], infos: [ { @@ -493,9 +459,62 @@ test('transcriptTask runs native transcript recipe through action bridge when av mimeType: 'audio/opus', index: 0, }, + { + key: 'blob-1-1', + url: 'https://affine.fail/api/copilot/blob/user-1/workspace-1/blob-1-1', + mimeType: 'audio/opus', + index: 1, + }, ], }); const bridgeInputs: unknown[] = []; + const clock = Sinon.useFakeTimers(); + t.teardown(() => clock.restore()); + const structuredCalls: { + messages: { content?: string; attachments?: unknown[] }[]; + options: { builtInRouteId?: string }; + slot?: string; + }[] = []; + let transientFailure = true; + const generateStructuredValue = Sinon.stub().callsFake( + async ( + _conditions: unknown, + messages: { content?: string; attachments?: unknown[] }[], + options: { builtInRouteId?: string }, + _contract: unknown, + _filter: unknown, + slot?: string + ) => { + structuredCalls.push({ messages, options, slot }); + if (options.builtInRouteId === 'Summarize the meeting structured') { + return { + value: { + title: 'Weekly Sync', + durationMinutes: 1, + attendees: ['A', 'B'], + keyPoints: ['Kickoff', 'Follow-up'], + actionItems: [], + decisions: [], + openQuestions: [], + blockers: [], + }, + }; + } + + const attachment = messages + .flatMap(message => message.attachments ?? []) + .at(0) as { attachment: string }; + if (attachment.attachment.includes('blob-1-1') && transientFailure) { + transientFailure = false; + throw new Error('upstream returned status 503: UNAVAILABLE'); + } + return { + value: attachment.attachment.includes('blob-1-0') + ? [{ a: 'A', s: 5, e: 9, t: 'Kickoff' }] + : [{ a: 'B', s: 100, e: 500, t: 'Follow-up' }], + }; + } + ); const claimDispatch = Sinon.stub(); claimDispatch.onFirstCall().resolves(true); claimDispatch.onSecondCall().resolves(false); @@ -519,20 +538,24 @@ test('transcriptTask runs native transcript recipe through action bridge when av } as never, {} as never, { - presignGet: Sinon.stub().resolves( - 'https://canary.copilotcontent.affine.pro/blob-1-0?sig=test' + presignGet: Sinon.stub().callsFake( + async (_userId, _workspaceId, key) => + `https://canary.copilotcontent.affine.pro/${key}?sig=test` ), } as never, {} as never, createTranscriptPromptService() as never, - createSuccessfulTranscriptBridge('run-bridge', bridgeInputs) as never + createSuccessfulTranscriptBridge('run-bridge', bridgeInputs) as never, + { generateStructuredValue } as never ); - await service.transcriptTask({ + const run = service.transcriptTask({ taskId: 'task-1', payload, generation: 'generation-1', }); + await clock.tickAsync(5_000); + await run; await service.transcriptTask({ taskId: 'task-1', payload, @@ -546,39 +569,63 @@ test('transcriptTask runs native transcript recipe through action bridge when av }); t.like((bridgeInputs[0] as { step: Record }).step, { slot: 'transcript.audio', - builtInRouteId: 'Transcript audio structured', + builtInRouteId: 'Transcript audio', }); t.deepEqual( ( bridgeInputs[0] as { - nativeInput: { input: { infos: unknown[] } }; + inputSnapshot: { infos: unknown[] }; } - ).nativeInput.input.infos, + ).inputSnapshot.infos, [ { url: 'https://canary.copilotcontent.affine.pro/blob-1-0?sig=test', mimeType: 'audio/opus', index: 0, }, + { + url: 'https://canary.copilotcontent.affine.pro/blob-1-1?sig=test', + mimeType: 'audio/opus', + index: 1, + }, ] ); - const messages = ( - bridgeInputs[0] as { - step: { - messages: { content?: string; attachments?: unknown[] }[]; - }; - } - ).step.messages; - t.false(messages[0].content?.includes('data:image/png')); - t.like(JSON.parse(messages[0].content ?? '{}'), { - infos: [{ mimeType: 'audio/opus', index: 0 }], - }); - t.deepEqual(messages.at(-1)?.attachments, [ - { - attachment: 'https://canary.copilotcontent.affine.pro/blob-1-0?sig=test', - mimeType: 'audio/opus', - }, - ]); + t.is(structuredCalls.length, 4); + const transcriptCalls = structuredCalls.filter( + call => call.options.builtInRouteId === 'Transcript audio' + ); + t.is(transcriptCalls.length, 3); + t.true(transcriptCalls.every(call => call.slot === 'transcript.audio')); + t.deepEqual( + transcriptCalls.map(call => call.messages.at(-1)?.attachments), + [ + [ + { + attachment: + 'https://canary.copilotcontent.affine.pro/blob-1-0?sig=test', + mimeType: 'audio/opus', + }, + ], + [ + { + attachment: + 'https://canary.copilotcontent.affine.pro/blob-1-1?sig=test', + mimeType: 'audio/opus', + }, + ], + [ + { + attachment: + 'https://canary.copilotcontent.affine.pro/blob-1-1?sig=test', + mimeType: 'audio/opus', + }, + ], + ] + ); + t.is( + structuredCalls.at(-1)?.messages.at(-1)?.content, + '00:00:17 A: Kickoff\n00:01:42 B: Follow-up' + ); t.like(completeDispatch.firstCall.args[3], { status: 'ready', errorCode: null, @@ -592,7 +639,16 @@ test('transcriptTask runs native transcript recipe through action bridge when av ); t.is( completeDispatch.firstCall.args[3].protectedResult.normalizedTranscript, - '00:00:05 A: Kickoff' + '00:00:17 A: Kickoff\n00:01:42 B: Follow-up' + ); + t.like( + completeDispatch.firstCall.args[3].protectedResult.normalizedSegments[1], + { + startSec: 102, + endSec: 342, + start: '00:01:42', + end: '00:05:42', + } ); t.deepEqual( completeDispatch.firstCall.args[3].protectedResult.infos, @@ -626,9 +682,9 @@ test('transcriptTask fails task when native action bridge reports an error event {} as never, createTranscriptPromptService() as never, { - runStream: (input: unknown) => + runStream: (input: any) => (async function* () { - await buildNativeTranscriptResult(input, 'run-bridge'); + await input.onRunCreated?.({ runId: 'run-bridge', attempt: 1 }); yield { type: 'error' as const, actionId: 'transcript.audio', diff --git a/packages/backend/server/src/plugins/copilot/runtime/action-runtime-bridge.ts b/packages/backend/server/src/plugins/copilot/runtime/action-runtime-bridge.ts index 2a29758d82..5f7ac8b80f 100644 --- a/packages/backend/server/src/plugins/copilot/runtime/action-runtime-bridge.ts +++ b/packages/backend/server/src/plugins/copilot/runtime/action-runtime-bridge.ts @@ -49,6 +49,15 @@ export type ActionRuntimeBridgeEvent = NativeActionEvent & { runId: string; }; +export type ActionRuntimeBridgeOutput = { + result: unknown; + attachments?: unknown[]; +}; + +export type ActionRuntimeBridgeExecutor = ( + input: ActionRuntimeBridgeInput +) => Promise; + export type ActionRuntimeBridgeRunContext = { runId: string; attempt: number; @@ -173,7 +182,8 @@ export class ActionRuntimeBridge { } async *runStream( - input: ActionRuntimeBridgeInput + input: ActionRuntimeBridgeInput, + executor?: ActionRuntimeBridgeExecutor ): AsyncIterableIterator { const attempt = await this.resolveAttempt(input); const run = await this.models.copilotActionRun.create({ @@ -206,8 +216,10 @@ export class ActionRuntimeBridge { status: 'running', }; yield { ...actionStart, runId: run.id }; - const output = await this.execute(inputWithBillingUnit); - for (const artifact of output.attachments) { + const output = executor + ? await executor(inputWithBillingUnit) + : await this.execute(inputWithBillingUnit); + for (const artifact of output.attachments ?? []) { const attachment = input.persistAttachment ? await input.persistAttachment(artifact) : artifact; diff --git a/packages/backend/server/src/plugins/copilot/transcript/constants.ts b/packages/backend/server/src/plugins/copilot/transcript/constants.ts index ebd6a4cd97..ba5573f166 100644 --- a/packages/backend/server/src/plugins/copilot/transcript/constants.ts +++ b/packages/backend/server/src/plugins/copilot/transcript/constants.ts @@ -1,3 +1,4 @@ export const TRANSCRIPT_ACTION_ID = 'transcript.audio'; -export const TRANSCRIPT_PROMPT_REF = 'Transcript audio structured'; +export const TRANSCRIPT_PROMPT_REF = 'Transcript audio'; +export const TRANSCRIPT_SUMMARY_PROMPT_REF = 'Summarize the meeting structured'; export const TRANSCRIPT_ACTION_VERSION = 'v1'; diff --git a/packages/backend/server/src/plugins/copilot/transcript/projection.ts b/packages/backend/server/src/plugins/copilot/transcript/projection.ts index a5fc13bf9b..c384d33269 100644 --- a/packages/backend/server/src/plugins/copilot/transcript/projection.ts +++ b/packages/backend/server/src/plugins/copilot/transcript/projection.ts @@ -6,6 +6,14 @@ import type { TranscriptionPayloadV2, } from './types'; +export type RawTranscriptSegment = { + sliceIndex: number; + speaker: string; + startSec: number; + endSec: number; + text: string; +}; + function formatSection(title: string, items: string[]) { if (!items.length) { return []; @@ -26,6 +34,58 @@ export function formatTranscriptTime(time: number) { .join(':'); } +export function normalizeTranscriptSegments( + rawSegments: RawTranscriptSegment[] +): NormalizedTranscriptSegment[] { + const normalized: NormalizedTranscriptSegment[] = []; + const dedupe = new Set(); + const sorted = [...rawSegments].sort((left, right) => { + return ( + left.startSec - right.startSec || + left.endSec - right.endSec || + left.sliceIndex - right.sliceIndex + ); + }); + + for (const segment of sorted) { + const text = segment.text.trim(); + if (!text) continue; + + const startSec = Math.max( + normalized.at(-1)?.endSec ?? 0, + segment.startSec, + 0 + ); + const endSec = Math.max(segment.endSec, startSec); + if (endSec <= startSec) continue; + + const speaker = segment.speaker.trim() || 'Speaker'; + const key = `${speaker}|${startSec}|${endSec}|${text}`; + if (dedupe.has(key)) continue; + + dedupe.add(key); + normalized.push({ + speaker, + startSec, + endSec, + start: formatTranscriptTime(startSec), + end: formatTranscriptTime(endSec), + text, + }); + } + + return normalized; +} + +export function buildNormalizedTranscript( + segments: NormalizedTranscriptSegment[] +) { + return segments + .map(segment => `${segment.start} ${segment.speaker}: ${segment.text}`) + .join('\n') + .trim(); +} + export function toLegacyTranscriptionSegments( segments: NormalizedTranscriptSegment[] ): LegacyTranscriptionSegment[] { diff --git a/packages/backend/server/src/plugins/copilot/transcript/schema.ts b/packages/backend/server/src/plugins/copilot/transcript/schema.ts index a1faeca1d2..af5cedb686 100644 --- a/packages/backend/server/src/plugins/copilot/transcript/schema.ts +++ b/packages/backend/server/src/plugins/copilot/transcript/schema.ts @@ -1,7 +1,9 @@ import { z } from 'zod'; -import { llmGetContractSchema } from '../../../native'; -import { buildStructuredResponseFromSchemaJson } from '../runtime/contracts'; +import { + buildStructuredResponseContract, + type RequiredStructuredOutputContract, +} from '../runtime/contracts'; // Owner: DB/job/API legacy compatibility and transcript projection. // Native owns transcript domain result schemas; this file accepts historical @@ -61,6 +63,15 @@ export const MeetingSummaryV2Schema = z.object({ blockers: z.array(z.string()), }); +export const TranscriptionResponseSchema = z.array( + z.object({ + a: z.string().describe("speaker's name, for example A, B, C"), + s: z.number().describe('start time in seconds'), + e: z.number().describe('end time in seconds'), + t: z.string().describe('transcription text'), + }) +); + export const TranscriptionSourceAudioSchema = z.object({ blobId: z.string().nullable().optional(), mimeType: z.string().nullable().optional(), @@ -103,8 +114,8 @@ export const TranscriptionSubmitInputSchema = TranscriptionPayloadV2Schema.pick( } ); -function buildRequiredStructuredContract(schema: Record) { - const contract = buildStructuredResponseFromSchemaJson(schema); +function buildRequiredStructuredContract(schema: z.ZodType) { + const contract = buildStructuredResponseContract(schema); if (!contract.responseSchemaJson || !contract.schemaHash) { throw new Error('Structured transcript contract is required'); } @@ -112,11 +123,15 @@ function buildRequiredStructuredContract(schema: Record) { return { responseSchemaJson: contract.responseSchemaJson, schemaHash: contract.schemaHash, - }; + } satisfies RequiredStructuredOutputContract; } -export const TranscriptActionResultContract = buildRequiredStructuredContract( - llmGetContractSchema('transcriptGeneratedResult') +export const TranscriptionResponseContract = buildRequiredStructuredContract( + TranscriptionResponseSchema +); + +export const MeetingSummaryV2Contract = buildRequiredStructuredContract( + MeetingSummaryV2Schema ); type CanonicalTranscriptPayload = z.infer; diff --git a/packages/backend/server/src/plugins/copilot/transcript/service.ts b/packages/backend/server/src/plugins/copilot/transcript/service.ts index e6badcf384..b8c6551d28 100644 --- a/packages/backend/server/src/plugins/copilot/transcript/service.ts +++ b/packages/backend/server/src/plugins/copilot/transcript/service.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'node:crypto'; +import { setTimeout } from 'node:timers/promises'; import { BadRequestException, Injectable } from '@nestjs/common'; import { AiJobStatus } from '@prisma/client'; @@ -16,27 +17,48 @@ import { } from '../../../core/realtime'; import { Models } from '../../../models'; import { PromptService } from '../prompt'; -import { ActionRuntimeBridge } from '../runtime/action-runtime-bridge'; +import type { + CopilotStructuredOptions, + PromptMessage, +} from '../providers/types'; +import { + ActionRuntimeBridge, + type ActionRuntimeBridgeInput, +} from '../runtime/action-runtime-bridge'; import { CapabilityRuntime } from '../runtime/capability-runtime'; +import type { RequiredStructuredOutputContract } from '../runtime/contracts'; import { CopilotStorage } from '../storage'; import { TRANSCRIPT_ACTION_ID, TRANSCRIPT_ACTION_VERSION, TRANSCRIPT_PROMPT_REF, + TRANSCRIPT_SUMMARY_PROMPT_REF, } from './constants'; import { taskToJob, type TranscriptionJob } from './job'; +import { + buildNormalizedTranscript, + normalizeTranscriptSegments, + type RawTranscriptSegment, +} from './projection'; import { CopilotTranscriptionRetryService } from './retry'; import { - TranscriptActionResultContract, + MeetingSummaryV2Contract, + MeetingSummaryV2Schema, + TranscriptionResponseContract, + TranscriptionResponseSchema, TranscriptPayloadSchema, } from './schema'; import type { + AudioBlobInfo, AudioBlobInfos, TranscriptionPayloadV2, TranscriptionSubmitInput, } from './types'; import { readStream } from './utils'; +const TRANSCRIPT_SLICE_CONCURRENCY = 2; +const TRANSCRIPT_RETRY_DELAYS = [5_000, 15_000]; + @Injectable() export class CopilotTranscriptionService { constructor( @@ -140,35 +162,223 @@ export class CopilotTranscriptionService { } satisfies TranscriptionPayloadV2; } - private async buildTranscriptActionMessages(payload: TranscriptionPayloadV2) { + private async buildTranscriptSliceMessages(info: AudioBlobInfo) { const prompt = await this.prompts.get(TRANSCRIPT_PROMPT_REF); if (!prompt) { - throw new Error('Transcript action prompt not found'); + throw new Error('Transcript prompt not found'); } - const metadata = { - sourceAudio: payload.sourceAudio ?? null, - quality: payload.quality ?? null, - sliceManifest: payload.sliceManifest ?? null, - infos: - payload.infos?.map(info => ({ - mimeType: info.mimeType, - index: info.index ?? null, - })) ?? null, - }; - const attachments = (payload.infos ?? []).map(info => ({ - role: 'user' as const, - content: `Audio attachment ${info.index ?? 0}`, - attachments: [{ attachment: info.url, mimeType: info.mimeType }], - params: { mimetype: info.mimeType }, - })); + return [ - ...this.prompts.finish(prompt, { - content: JSON.stringify(metadata), - }), - ...attachments, + ...this.prompts.finish(prompt, {}), + { + role: 'user' as const, + content: + 'Transcribe this audio slice. Return start and end timestamps as elapsed seconds relative to this slice; never encode MM:SS as a number.', + attachments: [{ attachment: info.url, mimeType: info.mimeType }], + params: { mimetype: info.mimeType }, + }, ]; } + private async buildMeetingSummaryMessages(normalizedTranscript: string) { + const prompt = await this.prompts.get(TRANSCRIPT_SUMMARY_PROMPT_REF); + if (!prompt) { + throw new Error('Transcript summary prompt not found'); + } + return this.prompts.finish(prompt, { content: normalizedTranscript }); + } + + private rebaseManifestlessSlices( + infos: AudioBlobInfos, + slices: RawTranscriptSegment[][] + ) { + let accumulatedOffset = 0; + return slices + .map((segments, fallbackIndex) => ({ + fallbackIndex, + sliceIndex: infos[fallbackIndex]?.index ?? fallbackIndex, + segments, + })) + .sort( + (left, right) => + left.sliceIndex - right.sliceIndex || + left.fallbackIndex - right.fallbackIndex + ) + .flatMap(({ segments }) => { + const rebased = segments.map(segment => ({ + ...segment, + startSec: segment.startSec + accumulatedOffset, + endSec: segment.endSec + accumulatedOffset, + })); + accumulatedOffset += Math.max( + 0, + ...segments.map(segment => segment.endSec) + ); + return rebased; + }); + } + + private async transcribeSlice( + input: ActionRuntimeBridgeInput, + info: AudioBlobInfo, + fallbackIndex: number, + offset: number, + durationSec?: number + ): Promise { + const messages = await this.buildTranscriptSliceMessages(info); + const output = await this.generateStructuredValue( + input, + messages, + TRANSCRIPT_PROMPT_REF, + TranscriptionResponseContract, + 'transcript.audio' + ); + const sliceIndex = info.index ?? fallbackIndex; + const response = TranscriptionResponseSchema.parse(output.value); + const timestamps = response.flatMap(segment => [segment.s, segment.e]); + const maxTimestamp = Math.max(0, ...timestamps); + const mmssTimestamps = timestamps.map(timestamp => { + const minutes = Math.floor(timestamp / 100); + const seconds = timestamp - minutes * 100; + return seconds < 60 ? minutes * 60 + seconds : null; + }); + const usesMmss = + durationSec !== undefined && + maxTimestamp > durationSec + 5 && + mmssTimestamps.every( + timestamp => timestamp !== null && timestamp <= durationSec + 5 + ); + if ( + durationSec !== undefined && + maxTimestamp > durationSec + 5 && + !usesMmss + ) { + throw new Error('Transcript slice timestamps exceed audio duration'); + } + + return response.map((segment, index) => { + const startSec = usesMmss + ? (mmssTimestamps[index * 2] ?? segment.s) + : segment.s; + const endSec = usesMmss + ? (mmssTimestamps[index * 2 + 1] ?? segment.e) + : segment.e; + return { + sliceIndex, + speaker: segment.a, + startSec: Math.min(startSec, durationSec ?? startSec) + offset, + endSec: Math.min(endSec, durationSec ?? endSec) + offset, + text: segment.t, + }; + }); + } + + private async generateStructuredValue( + input: ActionRuntimeBridgeInput, + messages: PromptMessage[], + builtInRouteId: string, + contract: RequiredStructuredOutputContract, + slot = 'prompt.structured' + ) { + for (let attempt = 0; ; attempt++) { + try { + return await this.runtime.generateStructuredValue( + { + profileId: input.step.profileId, + modelId: input.step.modelId, + }, + messages, + { + ...(input.step.options as CopilotStructuredOptions | undefined), + builtInRouteId, + }, + contract, + undefined, + slot + ); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + const retryable = + /upstream returned status (?:429|5\d\d)|RESOURCE_EXHAUSTED|UNAVAILABLE|llm_timeout|timed? out|fetch failed/i.test( + message + ); + const delay = TRANSCRIPT_RETRY_DELAYS[attempt]; + if (!retryable || delay === undefined || input.signal?.aborted) { + throw error; + } + await setTimeout(delay, undefined, { signal: input.signal }); + } + } + } + + private async executeTranscriptAction( + input: ActionRuntimeBridgeInput, + payload: TranscriptionPayloadV2 + ) { + const infos = payload.infos ?? []; + const slices: RawTranscriptSegment[][] = []; + const manifestProvided = !!payload.sliceManifest?.length; + + for ( + let batchStart = 0; + batchStart < infos.length; + batchStart += TRANSCRIPT_SLICE_CONCURRENCY + ) { + const batch = infos.slice( + batchStart, + batchStart + TRANSCRIPT_SLICE_CONCURRENCY + ); + await Promise.all( + batch.map(async (info, batchIndex) => { + const index = batchStart + batchIndex; + const manifestItem = manifestProvided + ? payload.sliceManifest?.find( + item => item.index === (info.index ?? index) + ) + : undefined; + slices[index] = await this.transcribeSlice( + input, + info, + index, + manifestItem?.startSec ?? 0, + manifestItem?.durationSec + ); + }) + ); + } + + const rawSegments = manifestProvided + ? slices.flat() + : this.rebaseManifestlessSlices(infos, slices); + const normalizedSegments = normalizeTranscriptSegments(rawSegments); + const normalizedTranscript = buildNormalizedTranscript(normalizedSegments); + let summaryJson = null; + + if (normalizedTranscript) { + const messages = + await this.buildMeetingSummaryMessages(normalizedTranscript); + const output = await this.generateStructuredValue( + input, + messages, + TRANSCRIPT_SUMMARY_PROMPT_REF, + MeetingSummaryV2Contract + ); + summaryJson = MeetingSummaryV2Schema.parse(output.value); + } + + return { + result: { + sourceAudio: payload.sourceAudio, + quality: payload.quality, + sliceManifest: payload.sliceManifest, + normalizedSegments, + normalizedTranscript, + summaryJson, + version: 'transcript-result-v1', + } satisfies TranscriptionPayloadV2, + }; + } + async submitTask( userId: string, workspaceId: string, @@ -306,46 +516,47 @@ export class CopilotTranscriptionService { task.workspaceId, payload ); - const messages = await this.buildTranscriptActionMessages(runtimePayload); - for await (const event of this.actionBridge.runStream({ - userId: task.userId, - workspaceId: task.workspaceId, - actionId: TRANSCRIPT_ACTION_ID, - actionVersion: TRANSCRIPT_ACTION_VERSION, - retryOf: retryOf ?? null, - inputSnapshot: runtimePayload, - onRunCreated: async ({ runId }) => { - const attached = - await this.models.copilotTranscriptTask.attachActionRun( + for await (const event of this.actionBridge.runStream( + { + userId: task.userId, + workspaceId: task.workspaceId, + actionId: TRANSCRIPT_ACTION_ID, + actionVersion: TRANSCRIPT_ACTION_VERSION, + retryOf: retryOf ?? null, + inputSnapshot: runtimePayload, + onRunCreated: async ({ runId }) => { + const attached = + await this.models.copilotTranscriptTask.attachActionRun( + taskId, + generation, + actionRunId, + runId + ); + if (!attached) { + throw new Error('stale transcript dispatch generation'); + } + actionRunId = runId; + this.publishTaskChanged( + task.workspaceId, taskId, - generation, - actionRunId, - runId + AiJobStatus.running ); - if (!attached) { - throw new Error('stale transcript dispatch generation'); - } - actionRunId = runId; - this.publishTaskChanged( - task.workspaceId, - taskId, - AiJobStatus.running - ); - }, - step: { - slot: 'transcript.audio', - builtInRouteId: TRANSCRIPT_PROMPT_REF, - messages, - options: { - user: task.userId, - workspace: task.workspaceId, - taskId, - billingUnitId: taskId, - featureKind: 'transcript', }, - responseContract: TranscriptActionResultContract, + step: { + slot: 'transcript.audio', + builtInRouteId: TRANSCRIPT_PROMPT_REF, + messages: [], + options: { + user: task.userId, + workspace: task.workspaceId, + taskId, + billingUnitId: taskId, + featureKind: 'transcript', + }, + }, }, - })) { + input => this.executeTranscriptAction(input, runtimePayload) + )) { if (event.type === 'error' || event.status === 'failed') { bridgeFailed = true; bridgeError = event.errorMessage ?? event.errorCode ?? bridgeError;