diff --git a/src/backend/drivers/ai-speech2speech/VoiceChangerDriver.test.ts b/src/backend/drivers/ai-speech2speech/VoiceChangerDriver.test.ts index 9a8c7cad1..f8ac1b3f3 100644 --- a/src/backend/drivers/ai-speech2speech/VoiceChangerDriver.test.ts +++ b/src/backend/drivers/ai-speech2speech/VoiceChangerDriver.test.ts @@ -47,7 +47,13 @@ import { PuterServer } from '../../server.js'; import type { MeteringService } from '../../services/metering/MeteringService.js'; import { setupTestServer } from '../../testUtil.js'; import { generateDefaultFsentries } from '../../util/userProvisioning.js'; -import type { VoiceChangerDriver } from './VoiceChangerDriver.js'; +import type { DriverStreamResult } from '../meta.js'; +import { + OggOpusClock, + outputBytesPerSecond, + type VoiceChangerDriver, + wavDataOffset, +} from './VoiceChangerDriver.js'; import { callerError, readStreamResult, @@ -330,11 +336,11 @@ describe('VoiceChangerDriver.convert success path', () => { it('meters one usage line at the per-second rate from costs.ts', async () => { const { actor } = await makeUser(); + // 32 KB of mp3_44100_128 back = 2 seconds. fetchSpy.mockResolvedValueOnce( - okResponse(new ArrayBuffer(8), 'audio/mpeg'), + okResponse(new ArrayBuffer(32_000), 'audio/mpeg'), ); - // 32 KB audio at 16 kbit/s = 2 seconds rounded up. const buf = Buffer.alloc(32_000); await readStreamResult( await withActor(actor, () => @@ -344,7 +350,7 @@ describe('VoiceChangerDriver.convert success path', () => { const usageType = 'elevenlabs:eleven_multilingual_sts_v2:second'; const perSecond = VOICE_CHANGER_COSTS[usageType]; - const expectedSeconds = Math.max(1, Math.ceil(32_000 / 16000)); + const expectedSeconds = 2; const calls = incrementUsageSpy.mock.calls.filter( ([, type]) => type === usageType, @@ -365,7 +371,7 @@ describe('VoiceChangerDriver.convert success path', () => { .map(([, , , cost]) => cost); const convert = async (charges: number) => { fetchSpy.mockResolvedValueOnce( - okResponse(new ArrayBuffer(8), 'audio/mpeg'), + okResponse(new ArrayBuffer(32_000), 'audio/mpeg'), ); await readStreamResult( await withActor(actor, () => @@ -385,6 +391,230 @@ describe('VoiceChangerDriver.convert success path', () => { expect(held).toEqual([base, base! * 2]); }); + it.each<[string, number]>([ + ['mp3_44100_128', 16_000], + ['mp3_22050_32', 4_000], + ['opus_48000_64', 8_000], + ['pcm_44100', 88_200], + ['ulaw_8000', 8_000], + ['wav_44100', 88_200], + ['mp3_44100_999999', 16_000], + ['flac_44100', 16_000], + ])('sizes %s at %s bytes per second', (format, bytes) => { + expect(outputBytesPerSecond(format)).toBe(bytes); + }); + + /** A 16-bit mono WAV header, with an optional LIST chunk before data. */ + const wavHeader = (sampleRate: number, dataBytes: number, list = 0) => { + const listChunk = list + ? Buffer.concat([ + Buffer.from('LIST'), + Buffer.from(new Uint32Array([list]).buffer), + Buffer.alloc(list), + ]) + : Buffer.alloc(0); + const fmt = Buffer.alloc(24); + fmt.write('fmt ', 0, 'ascii'); + fmt.writeUInt32LE(16, 4); + fmt.writeUInt16LE(1, 8); + fmt.writeUInt16LE(1, 10); + fmt.writeUInt32LE(sampleRate, 12); + fmt.writeUInt32LE(sampleRate * 2, 16); + fmt.writeUInt16LE(2, 20); + fmt.writeUInt16LE(16, 22); + const data = Buffer.alloc(8); + data.write('data', 0, 'ascii'); + data.writeUInt32LE(dataBytes, 4); + const body = Buffer.concat([fmt, listChunk, data]); + const riff = Buffer.alloc(12); + riff.write('RIFF', 0, 'ascii'); + riff.writeUInt32LE(4 + body.length + dataBytes, 4); + riff.write('WAVE', 8, 'ascii'); + return Buffer.concat([riff, body]); + }; + + it("finds where a WAV stream's samples start", () => { + expect(wavDataOffset(wavHeader(16_000, 0))).toBe(44); + expect(wavDataOffset(wavHeader(16_000, 0, 26))).toBe(78); + expect(wavDataOffset(wavHeader(16_000, 0).subarray(0, 40))).toBe( + undefined, + ); + expect(wavDataOffset(Buffer.alloc(64))).toBe(undefined); + }); + + /** One Ogg page carrying `body` as a single packet. */ + const oggPage = (granule: bigint, sequence: number, body: Buffer) => { + const lacing: number[] = []; + let left = body.length; + while (left >= 255) { + lacing.push(255); + left -= 255; + } + lacing.push(left); + const header = Buffer.alloc(27); + header.write('OggS', 0, 'ascii'); + header.writeBigUInt64LE(granule, 6); + header.writeUInt32LE(sequence, 18); + header[26] = lacing.length; + return Buffer.concat([header, Buffer.from(lacing), body]); + }; + + /** A sparse Ogg Opus stream: `seconds` of audio in very few bytes. */ + const oggOpus = (seconds: number, preSkip = 312) => { + const opusHead = Buffer.alloc(19); + opusHead.write('OpusHead', 0, 'ascii'); + opusHead[8] = 1; + opusHead[9] = 1; + opusHead.writeUInt16LE(preSkip, 10); + opusHead.writeUInt32LE(48_000, 12); + const pages = [ + oggPage(0n, 0, opusHead), + oggPage(0n, 1, Buffer.from('OpusTags')), + ]; + for (let s = 1; s <= seconds; s++) { + pages.push( + oggPage(BigInt(preSkip + s * 48_000), s + 1, Buffer.alloc(300)), + ); + } + return Buffer.concat(pages); + }; + + it('times an Ogg Opus stream by its last granule, however it is chunked', () => { + const stream = oggOpus(7); + const whole = new OggOpusClock(); + whole.push(stream); + expect(whole.seconds).toBe(7); + + const trickle = new OggOpusClock(); + for (const byte of stream) trickle.push(Buffer.from([byte])); + expect(trickle.seconds).toBe(7); + }); + + it('does not time a stream that is not Ogg Opus', () => { + const clock = new OggOpusClock(); + clock.push(Buffer.alloc(4096, 1)); + expect(clock.seconds).toBe(undefined); + }); + + describe('meters the length of the audio that comes back', () => { + const usageType = 'elevenlabs:eleven_multilingual_sts_v2:second'; + const billed = async () => { + await vi.waitFor(() => + expect( + incrementUsageSpy.mock.calls.filter( + ([, type]) => type === usageType, + ), + ).toHaveLength(1), + ); + const [, , count, cost] = incrementUsageSpy.mock.calls.find( + ([, type]) => type === usageType, + )!; + expect(cost).toBe(VOICE_CHANGER_COSTS[usageType]! * count); + return count; + }; + const convert = async ( + actor: Actor, + inputBytes: number, + args: Record = {}, + ) => + readStreamResult( + await withActor(actor, () => + driver.convert({ + audio: dataUrl(Buffer.alloc(inputBytes), 'audio/ogg'), + ...args, + }), + ), + ); + + it('bills 600 s for a compact 600 s clip, not its byte size', async () => { + const { actor } = await makeUser(); + // 600 s of Opus at 8 kbps in; 600 s of mp3_44100_128 out. + fetchSpy.mockResolvedValueOnce( + okResponse(new ArrayBuffer(600 * 16_000)), + ); + await convert(actor, 441_078); + expect(await billed()).toBe(600); + }); + + it('sizes the output by the requested output_format', async () => { + const { actor } = await makeUser(); + // pcm_16000 is 16-bit mono: 32,000 bytes per second. + fetchSpy.mockResolvedValueOnce( + okResponse(new ArrayBuffer(3 * 32_000), 'audio/pcm'), + ); + await convert(actor, 1_000_000, { output_format: 'pcm_16000' }); + expect(await billed()).toBe(3); + }); + + it('does not bill a WAV header as audio', async () => { + const { actor } = await makeUser(); + // 3 s of wav_16000 is 96,000 bytes of samples plus its header. + fetchSpy.mockResolvedValueOnce( + okResponse( + new Uint8Array( + Buffer.concat([ + wavHeader(16_000, 3 * 32_000, 26), + Buffer.alloc(3 * 32_000), + ]), + ).buffer, + 'audio/wav', + ), + ); + await convert(actor, 1_000_000, { output_format: 'wav_16000' }); + expect(await billed()).toBe(3); + }); + + it('bills sparse Opus output by its duration, not its bytes', async () => { + const { actor } = await makeUser(); + // 20 s of mostly silent opus_48000_64 is far under 20 s of bytes. + const body = oggOpus(20); + expect(body.length).toBeLessThan(8_000 * 2); + fetchSpy.mockResolvedValueOnce( + okResponse(new Uint8Array(body).buffer, 'audio/ogg'), + ); + await convert(actor, 1_000, { output_format: 'opus_48000_64' }); + expect(await billed()).toBe(20); + }); + + it('sizes an undocumented output_format as the default', async () => { + const { actor } = await makeUser(); + // 3 s at the default mp3_44100_128. + fetchSpy.mockResolvedValueOnce( + okResponse(new ArrayBuffer(3 * 16_000)), + ); + await convert(actor, 1_000_000, { + output_format: 'mp3_44100_999999', + }); + expect(await billed()).toBe(3); + }); + + it('charges at least the input estimate when the caller hangs up', async () => { + const { actor } = await makeUser(); + fetchSpy.mockResolvedValueOnce( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new Uint8Array(16_000)); + }, + }), + { status: 200, headers: { 'content-type': 'audio/mpeg' } }, + ), + ); + const result = (await withActor(actor, () => + driver.convert({ + audio: dataUrl(Buffer.alloc(80_000), 'audio/mpeg'), + }), + )) as DriverStreamResult; + for await (const _chunk of result.stream) break; + expect(await billed()).toBe(5); + await vi.waitFor(async () => + expect( + await server.stores.creditHold.outstanding(actor.user.uuid), + ).toBe(0), + ); + }); + }); + it('meters nothing when ElevenLabs breaks off before any audio', async () => { const { actor } = await makeUser(); fetchSpy.mockResolvedValueOnce( diff --git a/src/backend/drivers/ai-speech2speech/VoiceChangerDriver.ts b/src/backend/drivers/ai-speech2speech/VoiceChangerDriver.ts index 3f904219d..f1f01bcb5 100644 --- a/src/backend/drivers/ai-speech2speech/VoiceChangerDriver.ts +++ b/src/backend/drivers/ai-speech2speech/VoiceChangerDriver.ts @@ -17,9 +17,11 @@ * along with this program. If not, see . */ +import { pipeline, Readable, Transform } from 'node:stream'; import { Context } from '../../core/context.js'; import { HttpError } from '../../core/http/HttpError.js'; import type { DriverStreamResult } from '../meta.js'; +import { OUTPUT_FORMATS } from '../ai-tts/providers/elevenlabs/ElevenLabsTTSProvider.js'; import { PuterDriver } from '../types.js'; import { type AiMeteringService, @@ -198,10 +200,9 @@ export class VoiceChangerDriver extends PuterDriver { legacyCode: 'bad_request', }); - // Metering: estimate duration from file size if we don't parse metadata. - // 16 kbit/s is a safe lower bound for speech audio; pre-check credits - // before we hit the ElevenLabs API. Post-usage we increment by the same - // estimate — duration parsing is deferred to v2.1 if needed. + // The hold estimates seconds from the input at 128 kbps: compact audio + // is under-held rather than refused. Usage is metered from the length + // of the audio that comes back. const estimatedSeconds = Math.max( 1, Math.ceil(loaded.buffer.byteLength / 16000), @@ -263,6 +264,10 @@ export class VoiceChangerDriver extends PuterDriver { if (search) url.search = search; const metering = this.#aiMetering; + const output = measuredOutput( + outputBytesPerSecond(outputFormat), + outputFormat, + ); return withStreamedAiCharge( metering, actor, @@ -287,16 +292,206 @@ export class VoiceChangerDriver extends PuterDriver { dataType: 'stream', content_type: response.headers.get('content-type') ?? 'audio/mpeg', - stream: upstreamBodyStream(response), + stream: output.measure(upstreamBodyStream(response)), }; }, - () => - metering.incrementUsage( + async () => { + const measured = await output.seconds; + // A stream cut short was still converted in full upstream. + const seconds = + measured.complete && measured.seconds + ? measured.seconds + : Math.max(measured.seconds ?? 0, estimatedSeconds); + return metering.incrementUsage( actor, usageKey, - estimatedSeconds, - estimatedCost, - ), + seconds, + ucentsPerSecond * seconds, + ); + }, ); } } + +/** + * Bytes per second of an ElevenLabs `output_format`. A format outside the + * documented set is sized as the default, so a made-up bitrate can't shrink the + * bill. + */ +export function outputBytesPerSecond(format: string): number { + const known = OUTPUT_FORMATS.has(format) ? format : DEFAULT_OUTPUT_FORMAT; + const [codec, rate, kbps] = known.split('_'); + if (codec === 'mp3' || codec === 'opus') return Number(kbps) * 125; + if (codec === 'ulaw' || codec === 'alaw') return Number(rate); + // pcm and wav: 16-bit mono. + return Number(rate) * 2; +} + +// Past this, a WAV whose data chunk hasn't started is billed by raw length. +const WAV_HEADER_MAX_BYTES = 64 * 1024; + +/** + * Where the samples of a RIFF/WAVE stream start, once `head` reaches its `data` + * chunk; undefined until then, or if `head` isn't a WAV. + */ +export function wavDataOffset(head: Buffer): number | undefined { + if (head.length < 12) return undefined; + if ( + head.toString('ascii', 0, 4) !== 'RIFF' || + head.toString('ascii', 8, 12) !== 'WAVE' + ) + return undefined; + let offset = 12; + while (offset + 8 <= head.length) { + if (head.toString('ascii', offset, offset + 4) === 'data') + return offset + 8; + const size = head.readUInt32LE(offset + 4); + offset += 8 + size + (size & 1); + } + return undefined; +} + +const OGG_HEADER_BYTES = 27; +const NO_GRANULE = 2n ** 64n - 1n; + +/** + * Walks an Ogg Opus stream page by page as it arrives. Opus is variable + * bitrate, so its length comes from the last page's granule position (48 kHz + * samples, less the stream's pre-skip) rather than its byte count. + */ +export class OggOpusClock { + #header = Buffer.alloc(0); + #skip = 0; + #firstBody: Buffer | undefined = Buffer.alloc(0); + #firstBodyBytes = 0; + #granule: bigint | undefined; + #preSkip = 0; + #broken = false; + + push(chunk: Buffer): void { + let offset = 0; + while (!this.#broken && offset < chunk.length) { + if (this.#skip > 0) { + const n = Math.min(this.#skip, chunk.length - offset); + if (this.#firstBody && this.#firstBody.length < 19) { + this.#firstBody = Buffer.concat([ + this.#firstBody, + chunk.subarray(offset, offset + n), + ]); + } + this.#skip -= n; + offset += n; + if (this.#skip === 0 && this.#firstBody) this.#readOpusHead(); + continue; + } + const want = + this.#header.length < OGG_HEADER_BYTES + ? OGG_HEADER_BYTES + : OGG_HEADER_BYTES + this.#header[26]!; + const n = Math.min( + want - this.#header.length, + chunk.length - offset, + ); + this.#header = Buffer.concat([ + this.#header, + chunk.subarray(offset, offset + n), + ]); + offset += n; + if (this.#header.length < want) continue; + if (this.#header.toString('ascii', 0, 4) !== 'OggS') { + this.#broken = true; + return; + } + if (want === OGG_HEADER_BYTES && this.#header[26]! > 0) continue; + const granule = this.#header.readBigUInt64LE(6); + if (granule !== NO_GRANULE) this.#granule = granule; + let body = 0; + for (const lacing of this.#header.subarray(OGG_HEADER_BYTES)) { + body += lacing; + } + this.#header = Buffer.alloc(0); + this.#skip = body; + if (this.#firstBody) this.#firstBodyBytes = body; + } + } + + /** Whole seconds so far, once the stream has proven to be Ogg Opus. */ + get seconds(): number | undefined { + if (this.#broken || this.#granule === undefined) return undefined; + const samples = Number(this.#granule) - this.#preSkip; + return samples > 0 + ? Math.max(1, Math.ceil(samples / 48_000)) + : undefined; + } + + #readOpusHead(): void { + const head = this.#firstBody!; + this.#firstBody = undefined; + if ( + this.#firstBodyBytes < 19 || + head.toString('ascii', 0, 8) !== 'OpusHead' + ) { + this.#broken = true; + return; + } + this.#preSkip = head.readUInt16LE(10); + } +} + +/** + * Counts the bytes of a converted stream. `seconds` settles when the stream + * ends or closes: its length in whole seconds (undefined when nothing came + * back), and whether the upstream sent all of it. A WAV's header isn't audio, + * so it isn't counted; Opus is timed by its pages. + */ +function measuredOutput(bytesPerSecond: number, format: string) { + let bytes = 0; + let head: Buffer | undefined = format.startsWith('wav_') + ? Buffer.alloc(0) + : undefined; + const opus = format.startsWith('opus_') ? new OggOpusClock() : undefined; + let headerBytes = 0; + let settle!: (measured: { seconds?: number; complete: boolean }) => void; + const seconds = new Promise<{ seconds?: number; complete: boolean }>( + (resolve) => { + settle = resolve; + }, + ); + const lengthOf = () => { + const timed = opus?.seconds; + if (timed !== undefined) return timed; + const audio = bytes - headerBytes; + return audio > 0 + ? Math.max(1, Math.ceil(audio / bytesPerSecond)) + : undefined; + }; + const measure = (source: Readable): Readable => { + const counter = new Transform({ + transform(chunk: Buffer, _encoding, callback) { + bytes += chunk.length; + opus?.push(chunk); + if (head) { + head = Buffer.concat([head, chunk]); + const offset = wavDataOffset(head); + if (offset !== undefined) headerBytes = offset; + if ( + offset !== undefined || + head.length >= WAV_HEADER_MAX_BYTES + ) + head = undefined; + } + callback(null, chunk); + }, + flush(callback) { + settle({ seconds: lengthOf(), complete: true }); + callback(); + }, + }); + counter.once('close', () => + settle({ seconds: lengthOf(), complete: false }), + ); + pipeline(source, counter, () => {}); + return counter; + }; + return { measure, seconds }; +} diff --git a/src/backend/drivers/ai-speech2txt/SpeechToTextDriver.test.ts b/src/backend/drivers/ai-speech2txt/SpeechToTextDriver.test.ts index 03c7a3c88..c69c3e0fb 100644 --- a/src/backend/drivers/ai-speech2txt/SpeechToTextDriver.test.ts +++ b/src/backend/drivers/ai-speech2txt/SpeechToTextDriver.test.ts @@ -508,6 +508,26 @@ describe('SpeechToTextDriver model selection and payload shape', () => { expect(sent.temperature).toBe(0.2); }); + it('keeps extra_body from overriding the response shape billing reads', async () => { + const { actor } = await makeUser(); + transcriptionsCreateMock.mockResolvedValueOnce({ text: 'x' }); + + await withActor(actor, () => + driver.transcribe({ + file: dataUrl(Buffer.from('a'), 'audio/mp3'), + model: 'gpt-transcribe', + extra_body: { + response_format: 'text', + stream: true, + chunking_strategy: 'auto', + }, + }), + ); + + const sent = transcriptionsCreateMock.mock.calls[0]![0]; + expect(sent.extra_body).toEqual({ chunking_strategy: 'auto' }); + }); + it('does not forward timestamp_granularities, which gpt-transcribe ignores', async () => { const { actor } = await makeUser(); transcriptionsCreateMock.mockResolvedValueOnce({ text: 'x' }); @@ -684,6 +704,72 @@ describe('SpeechToTextDriver metering', () => { expect(creditCall[1]).toBe(expected); }); + it('meters the duration OpenAI reports, not the byte estimate', async () => { + const { actor } = await makeUser(); + const usageType = 'openai:gpt-transcribe:second'; + const billed = async (bytes: number, seconds: number) => { + incrementUsageSpy.mockClear(); + transcriptionsCreateMock.mockResolvedValueOnce({ + text: 'x', + usage: { type: 'duration', seconds }, + }); + await withActor(actor, () => + driver.transcribe({ + file: dataUrl(Buffer.alloc(bytes, 1), 'audio/ogg'), + }), + ); + const [, , count, cost] = incrementUsageSpy.mock.calls.find( + ([, type]) => type === usageType, + )!; + expect(cost).toBe(SPEECH_TO_TEXT_COSTS[usageType]! * count); + return count; + }; + // 600 s of Opus at 8 kbps, and 60 s of 44.1 kHz stereo WAV. + expect(await billed(441_078, 600)).toBe(600); + expect(await billed(10_584_078, 59.2)).toBe(60); + }); + + it('asks for json behind response_format=text so usage comes back', async () => { + const { actor } = await makeUser(); + transcriptionsCreateMock.mockResolvedValueOnce({ + text: 'plain transcript', + usage: { type: 'duration', seconds: 42 }, + }); + const result = await withActor(actor, () => + driver.transcribe({ + file: dataUrl(Buffer.from('a'), 'audio/mp3'), + response_format: 'text', + }), + ); + expect(result).toBe('plain transcript'); + expect(transcriptionsCreateMock.mock.calls[0]![0].response_format).toBe( + 'json', + ); + const usageType = 'openai:gpt-transcribe:second'; + const [, , count] = incrementUsageSpy.mock.calls.find( + ([, type]) => type === usageType, + )!; + expect(count).toBe(42); + }); + + it('falls back to the byte estimate when OpenAI reports token usage', async () => { + const { actor } = await makeUser(); + transcriptionsCreateMock.mockResolvedValueOnce({ + text: 'x', + usage: { type: 'tokens', input_tokens: 10, output_tokens: 2 }, + }); + await withActor(actor, () => + driver.transcribe({ + file: dataUrl(Buffer.alloc(32000, 0), 'audio/mp3'), + }), + ); + const usageType = 'openai:gpt-transcribe:second'; + const [, , count] = incrementUsageSpy.mock.calls.find( + ([, type]) => type === usageType, + )!; + expect(count).toBe(2); + }); + it('holds and meters at the AI cost factor', async () => { const { actor } = await makeUser(); const usageType = 'openai:gpt-transcribe:second'; diff --git a/src/backend/drivers/ai-speech2txt/providers/openai/OpenAISpeechToTextProvider.ts b/src/backend/drivers/ai-speech2txt/providers/openai/OpenAISpeechToTextProvider.ts index d226302a9..0526c9b64 100644 --- a/src/backend/drivers/ai-speech2txt/providers/openai/OpenAISpeechToTextProvider.ts +++ b/src/backend/drivers/ai-speech2txt/providers/openai/OpenAISpeechToTextProvider.ts @@ -170,9 +170,9 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider { ); } - // Estimate seconds from raw bytes — 16 kbps is a conservative speech-audio - // lower bound. Full metadata parsing (music-metadata) is deferred — clients - // aren't observably sensitive to billing-time delta vs real duration. + // The hold estimates seconds from bytes at 128 kbps: compact audio is + // under-held rather than refused, and usage is metered from the + // duration OpenAI reports. const estimatedSeconds = Math.max( 1, Math.ceil(loaded.buffer.byteLength / 16000), @@ -198,8 +198,13 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider { file: openaiFile, model: selectedModel, }; + // A `text` response carries no usage, so json is asked for and + // its text returned. if (args.response_format) - payload.response_format = args.response_format; + payload.response_format = + args.response_format === 'text' + ? 'json' + : args.response_format; if (args.language) payload.language = args.language; if (typeof args.temperature === 'number') payload.temperature = args.temperature; @@ -207,7 +212,15 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider { // The API ignores a bare `logprobs` flag; it only honors `include`. if (args.logprobs && caps.canLogprobs) payload.include = ['logprobs']; - if (args.extra_body) payload.extra_body = args.extra_body; + if (args.extra_body && typeof args.extra_body === 'object') { + // Billing reads usage off the response; these can strip it. + const extra = { + ...(args.extra_body as Record), + }; + delete extra.response_format; + delete extra.stream; + payload.extra_body = extra; + } const result = await this.#openai.audio.transcriptions.create( payload as unknown as Parameters< @@ -215,11 +228,12 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider { >[0], ); + const billedSeconds = reportedSeconds(result) ?? estimatedSeconds; this.deps.metering.incrementUsage( actor, usageType, - estimatedSeconds, - ucentsPerSecond * estimatedSeconds, + billedSeconds, + ucentsPerSecond * billedSeconds, ); // Text response_format: return raw string; otherwise forward the OpenAI object. @@ -234,3 +248,18 @@ export class OpenAISpeechToTextProvider extends SpeechToTextProvider { } } } + +/** Whole seconds of audio OpenAI reports billing for, if it reports them. */ +function reportedSeconds(result: unknown): number | undefined { + if (!result || typeof result !== 'object') return undefined; + const { usage, duration } = result as { + usage?: { type?: string; seconds?: unknown }; + duration?: unknown; + }; + const seconds = usage?.type === 'duration' ? usage.seconds : duration; + return typeof seconds === 'number' && + Number.isFinite(seconds) && + seconds > 0 + ? Math.max(1, Math.ceil(seconds)) + : undefined; +} diff --git a/src/backend/drivers/ai-tts/providers/elevenlabs/ElevenLabsTTSProvider.ts b/src/backend/drivers/ai-tts/providers/elevenlabs/ElevenLabsTTSProvider.ts index a95e8f05a..911d3a2d0 100644 --- a/src/backend/drivers/ai-tts/providers/elevenlabs/ElevenLabsTTSProvider.ts +++ b/src/backend/drivers/ai-tts/providers/elevenlabs/ElevenLabsTTSProvider.ts @@ -48,7 +48,7 @@ const VOICE_ID_PATTERN = /^[A-Za-z0-9]{20}$/; const VOICE_SEARCH_MAX_PAGES = 5; // https://elevenlabs.io/docs/api-reference/text-to-speech/convert // mp3_44100_192 needs the Creator tier; pcm/wav at 44.1 kHz need Pro. -const OUTPUT_FORMATS = new Set([ +export const OUTPUT_FORMATS = new Set([ 'alaw_8000', 'mp3_22050_32', 'mp3_24000_48', diff --git a/src/backend/drivers/util/aiCostFactor.test.ts b/src/backend/drivers/util/aiCostFactor.test.ts index 407583746..874ff081b 100644 --- a/src/backend/drivers/util/aiCostFactor.test.ts +++ b/src/backend/drivers/util/aiCostFactor.test.ts @@ -651,6 +651,122 @@ describe('AI cost factor', () => { } }); + it('renews the hold until a slow charge lands', async () => { + vi.useFakeTimers({ toFake: ['setInterval', 'clearInterval'] }); + const refresh = vi.spyOn(server.stores.creditHold, 'refresh'); + const check = vi + .spyOn(metering, 'hasEnoughCredits') + .mockResolvedValue(true); + let landCharge = () => {}; + const charge = new Promise((resolve) => { + landCharge = resolve; + }); + try { + const result = await withStreamedAiCharge( + scoped, + actor, + usageType, + 1000, + async () => ({ + dataType: 'stream', + content_type: 'audio/mpeg', + stream: Readable.from([Buffer.from('ab')]), + }), + () => charge, + ); + expect(await read(result.stream)).toBeUndefined(); + + await vi.advanceTimersByTimeAsync(11 * 60 * 1000); + expect(refresh).toHaveBeenCalledTimes(2); + expect(await metering.getOutstandingHolds(actor)).toBe(1000); + + landCharge(); + await vi.waitFor(async () => + expect(await metering.getOutstandingHolds(actor)).toBe(0), + ); + await vi.advanceTimersByTimeAsync(10 * 60 * 1000); + expect(refresh).toHaveBeenCalledTimes(2); + } finally { + check.mockRestore(); + refresh.mockRestore(); + vi.useRealTimers(); + } + }); + + it('renews the hold while a slow upstream is still opening', async () => { + vi.useFakeTimers({ toFake: ['setInterval', 'clearInterval'] }); + const refresh = vi.spyOn(server.stores.creditHold, 'refresh'); + const check = vi + .spyOn(metering, 'hasEnoughCredits') + .mockResolvedValue(true); + let opened = () => {}; + const opening = new Promise((resolve) => { + opened = resolve; + }); + try { + const pending = withStreamedAiCharge( + scoped, + actor, + usageType, + 1000, + async () => { + await opening; + return { + dataType: 'stream', + content_type: 'audio/mpeg', + stream: Readable.from([Buffer.from('ab')]), + }; + }, + () => undefined, + ); + await vi.waitFor(async () => + expect(await metering.getOutstandingHolds(actor)).toBe( + 1000, + ), + ); + await vi.advanceTimersByTimeAsync(9 * 60 * 1000); + expect(refresh).toHaveBeenCalledTimes(1); + + opened(); + expect(await read((await pending).stream)).toBeUndefined(); + await vi.waitFor(async () => + expect(await metering.getOutstandingHolds(actor)).toBe(0), + ); + } finally { + check.mockRestore(); + refresh.mockRestore(); + vi.useRealTimers(); + } + }); + + it('releases the hold when recording the charge throws', async () => { + const check = vi + .spyOn(metering, 'hasEnoughCredits') + .mockResolvedValue(true); + try { + const result = await withStreamedAiCharge( + scoped, + actor, + usageType, + 1000, + async () => ({ + dataType: 'stream', + content_type: 'audio/mpeg', + stream: Readable.from([Buffer.from('ab')]), + }), + () => { + throw new Error('metering down'); + }, + ); + expect(await read(result.stream)).toBeUndefined(); + await vi.waitFor(async () => + expect(await metering.getOutstandingHolds(actor)).toBe(0), + ); + } finally { + check.mockRestore(); + } + }); + it('charges nothing when the upstream fails before its first byte', async () => { const { record, stream } = await relay( new Readable({ diff --git a/src/backend/drivers/util/aiCostFactor.ts b/src/backend/drivers/util/aiCostFactor.ts index 9a65b86a2..21603adf3 100644 --- a/src/backend/drivers/util/aiCostFactor.ts +++ b/src/backend/drivers/util/aiCostFactor.ts @@ -416,10 +416,14 @@ export async function withAiCreditHold( } } +// Under the hold store's default TTL, so a renewal always lands first. +const STREAM_HOLD_RENEW_MS = 5 * 60 * 1000; + /** * Holds `amount` while `open` starts an upstream audio stream, then charges it * through `record` once the first byte arrives, or once the caller hangs up. An - * upstream that fails or ends before its first byte is not charged. + * upstream that fails or ends before its first byte is not charged. The hold is + * renewed until the charge lands, however long a slow reader takes. */ export async function withStreamedAiCharge( metering: AiMeteringService, @@ -431,10 +435,20 @@ export async function withStreamedAiCharge( ): Promise { const hold = await metering.reserveAiCredits(actor, usageType, amount); if (!hold) throw insufficientCreditsError(); + // Started before `open`, which can itself run as long as the hold's TTL. + const renew = setInterval(async () => { + try { + await hold.extend?.(); + } catch { + // The next tick retries. + } + }, STREAM_HOLD_RENEW_MS); + renew.unref?.(); let result: DriverStreamResult; try { result = await open(); } catch (e) { + clearInterval(renew); await hold.release(); throw e; } @@ -445,11 +459,17 @@ export async function withStreamedAiCharge( settled = true; // The relay runs outside the request, so the hold waits on the charge // itself. A failed release leaves the hold to its TTL. - const release = () => hold.release().catch(() => {}); - void Promise.resolve(charge ? record() : undefined).then( - release, - release, - ); + const release = async () => { + clearInterval(renew); + try { + await hold.release(); + } catch { + // ignored + } + }; + void Promise.resolve() + .then(() => (charge ? record() : undefined)) + .then(release, release); }; const source = result.stream; let chunks: AsyncIterator | undefined; diff --git a/src/backend/services/metering/MeteringService.test.ts b/src/backend/services/metering/MeteringService.test.ts index d18e692f4..e4c92d3c5 100644 --- a/src/backend/services/metering/MeteringService.test.ts +++ b/src/backend/services/metering/MeteringService.test.ts @@ -2264,6 +2264,28 @@ describe('MeteringService', () => { expect(allowed.remaining).toBe(allowed.monthUsageAllowance + 5_000); }); + it('reads a spent addons record once across gate checks', async () => { + const addonsKey = `${POLICY_PREFIX}:actor:${actor.user.uuid}:addons`; + const getSpy = vi.spyOn(server.stores.kv, 'get'); + try { + for (let i = 0; i < 3; i++) await target.getAllowedUsage(actor); + const addonReads = getSpy.mock.calls.filter( + ([{ key }]) => key === addonsKey, + ); + expect(addonReads).toHaveLength(1); + } finally { + getSpy.mockRestore(); + } + }); + + it('sees credit granted after a gate check memoized the spent record', async () => { + const before = await target.getAllowedUsage(actor); + expect(before.addons).toEqual({}); + await target.updateAddonCredit(actor.user.uuid!, 5_000); + const after = await target.getAllowedUsage(actor); + expect(after.remaining).toBe(after.monthUsageAllowance + 5_000); + }); + it('clamps remaining at zero when over allowance with no credits', async () => { const sub = await target.getActorSubscription(actor); await target.incrementUsage( diff --git a/src/backend/services/metering/MeteringService.ts b/src/backend/services/metering/MeteringService.ts index defd89277..130f3b9b8 100644 --- a/src/backend/services/metering/MeteringService.ts +++ b/src/backend/services/metering/MeteringService.ts @@ -1578,7 +1578,7 @@ export class MeteringService extends PuterService { }> { const [userSubscription, addons, totals] = await Promise.all([ this.getActorSubscription(actor), - this.getActorAddons(actor), + this.#addonsForSettle(actor), this.currentMonthTotals(actor), ]); @@ -2244,7 +2244,7 @@ export class MeteringService extends PuterService { : null; } - /** `getActorAddons`, through `#spentAddons`. */ + /** `getActorAddons`, through `#spentAddons`. Shared by gates and settles. */ async #addonsForSettle(actor: Actor): Promise { const uuid = actor.user?.uuid; const cached = uuid ? this.#spentAddons.get(uuid) : undefined;