diff --git a/.changeset/livekit-audio-pipeline.md b/.changeset/livekit-audio-pipeline.md new file mode 100644 index 000000000..2b6e2104b --- /dev/null +++ b/.changeset/livekit-audio-pipeline.md @@ -0,0 +1,5 @@ +--- +"braintrust": patch +--- + +refactor(livekit): Record audio through shared audio pipeline diff --git a/js/src/audio-middleware.test.ts b/js/src/audio-middleware.test.ts new file mode 100644 index 000000000..d2a0ea219 --- /dev/null +++ b/js/src/audio-middleware.test.ts @@ -0,0 +1,194 @@ +import { expect, test, vi } from "vitest"; +import { + runMiddleware, + type AudioMiddleware, + type AudioSegmentMeta, +} from "./audio-middleware"; + +const meta: AudioSegmentMeta = { + index: 3, + startOffsetMs: 1000, + durationMs: 2, + sampleRate: 1000, + channels: [{ sources: ["a"] }, { sources: ["b"] }], + mimeType: "audio/pcm", + encoding: "pcm", + attributes: { + "pcm.sample_format": "s16le", + "pcm.channel_layout": "interleaved", + }, +}; +/** Two stereo frames: a = [2, 4], b = [6, 8]. */ +const pcm = () => new Uint8Array(Int16Array.from([2, 6, 4, 8]).buffer); +const samples = (data: Uint8Array) => + Array.from(new Int16Array(data.slice().buffer)); + +test("without middleware, the segment and metadata come back unchanged", async () => { + const result = await runMiddleware([], pcm(), meta); + expect(samples(result.segment)).toEqual([2, 6, 4, 8]); + expect(result.meta).toEqual(meta); +}); + +test("middleware runs outermost first, transforms input and observes results on the way back", async () => { + const calls: string[] = []; + const outer: AudioMiddleware = async (segment, meta, next) => { + calls.push(`outer in ${meta.encoding}`); + const result = await next(segment, meta); + calls.push(`outer out ${result.meta.encoding}`); + return result; + }; + const reverse: AudioMiddleware = async (segment, meta, next) => { + calls.push(`reverse in ${meta.encoding}`); + const result = await next(segment.slice().reverse(), { + ...meta, + encoding: "reversed", + mimeType: "application/octet-stream", + }); + calls.push(`reverse out ${result.meta.encoding}`); + return result; + }; + const result = await runMiddleware([outer, reverse], pcm(), meta); + expect(calls).toEqual([ + "outer in pcm", + "reverse in pcm", + "reverse out reversed", + "outer out reversed", + ]); + expect([...result.segment]).toEqual([...pcm()].reverse()); + expect(result.meta).toMatchObject({ + index: 3, + startOffsetMs: 1000, + encoding: "reversed", + mimeType: "application/octet-stream", + }); +}); + +test("middleware can short-circuit the rest of the stack", async () => { + const skipped = vi.fn(); + const replace: AudioMiddleware = (_segment, meta) => ({ + segment: Uint8Array.of(7), + meta: { ...meta, encoding: "replaced" }, + }); + const result = await runMiddleware([replace, skipped], pcm(), meta); + expect(skipped).not.toHaveBeenCalled(); + expect([...result.segment]).toEqual([7]); + expect(result.meta.encoding).toBe("replaced"); +}); + +test("attributes and state pass custom values to later middleware and the result", async () => { + let seen: unknown[] = []; + const tag: AudioMiddleware = (segment, meta, next) => + next(segment, { + ...meta, + attributes: { ...meta.attributes, "acme.key_id": "k-1" }, + state: { ...meta.state, "acme.key": Uint8Array.of(1, 2) }, + }); + const read: AudioMiddleware = (segment, meta, next) => { + seen = [meta.attributes?.["acme.key_id"], meta.state?.["acme.key"]]; + return next(segment, meta); + }; + const result = await runMiddleware([tag, read], pcm(), meta); + expect(seen).toEqual(["k-1", Uint8Array.of(1, 2)]); + expect(result.meta).toMatchObject({ + attributes: { "acme.key_id": "k-1" }, + state: { "acme.key": Uint8Array.of(1, 2) }, + }); +}); + +test("segments passed down may take any form; the stack must end with bytes", async () => { + type Words = { words: string[] }; + const split: AudioMiddleware = (segment, meta, next) => + next({ words: ["two", "frames", String(segment.length)] }, meta); + const join: AudioMiddleware = (segment, meta, next) => + next(new TextEncoder().encode(segment.words.join(" ")), { + ...meta, + mimeType: "text/plain", + }); + const result = await runMiddleware([split, join], pcm(), meta); + expect(new TextDecoder().decode(result.segment)).toBe("two frames 8"); + await expect(runMiddleware([split], pcm(), meta)).rejects.toThrow( + "invalid_audio_middleware_result", + ); +}); + +test("middleware that changes the layout describes it in the metadata", async () => { + // Averages the two channels into one, keeping every source on channel 0. + const mixdown: AudioMiddleware = (segment, meta, next) => { + const stereo = new Int16Array(segment.slice().buffer); + const mono = Int16Array.from( + { length: stereo.length / 2 }, + (_, i) => (stereo[i * 2] + stereo[i * 2 + 1]) / 2, + ); + return next(new Uint8Array(mono.buffer), { + ...meta, + channels: [{ sources: meta.channels.flatMap((c) => c.sources) }], + }); + }; + const result = await runMiddleware([mixdown], pcm(), meta); + expect(samples(result.segment)).toEqual([4, 6]); + expect(result.meta.channels).toEqual([{ sources: ["a", "b"] }]); +}); + +test.each<[string, AudioMiddleware]>([ + [ + "throws", + () => { + throw new Error("boom"); + }, + ], + [ + "rejects", + async () => { + throw new Error("boom"); + }, + ], + [ + "mutates metadata", + (segment, meta, next) => { + (meta as { encoding: string }).encoding = "changed"; + return next(segment, meta); + }, + ], + [ + "calls next twice", + async (segment, meta, next) => { + await next(segment, meta); + return next(segment, meta); + }, + ], + [ + "changes timeline placement", + (segment, meta, next) => + next(segment, { ...meta, startOffsetMs: meta.startOffsetMs + 1 }), + ], + [ + "changes segment identity", + async (segment, meta) => ({ + segment, + meta: { ...meta, index: meta.index + 1 }, + }), + ], + [ + "returns a non-byte segment", + async (_segment, meta) => ({ + segment: "pcm" as unknown as Uint8Array, + meta, + }), + ], + ["returns nothing", (async () => undefined) as unknown as AudioMiddleware], +])("middleware that %s rejects the run", async (_name, failing) => { + await expect(runMiddleware([failing], pcm(), meta)).rejects.toThrow(); +}); + +test("downstream failures reach middleware that ignores next() without unhandled rejections", async () => { + const detached: AudioMiddleware = (segment, meta, next) => { + void next(segment, meta); + return { segment, meta }; + }; + const failing: AudioMiddleware = () => { + throw new Error("boom"); + }; + const result = await runMiddleware([detached, failing], pcm(), meta); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(samples(result.segment)).toEqual([2, 6, 4, 8]); +}); diff --git a/js/src/audio-middleware.ts b/js/src/audio-middleware.ts new file mode 100644 index 000000000..e5cf5f6b2 --- /dev/null +++ b/js/src/audio-middleware.ts @@ -0,0 +1,147 @@ +import { isObject } from "../util"; + +/** + * Describes what a segment is and where it sits on the recording timeline, in + * terms any audio format shares. How it is encoded belongs in the segment. + */ +export interface AudioSegmentMeta { + /** Zero-based position of this segment within its recording. */ + readonly index: number; + /** Segment start on the recording timeline. Transforms must preserve it. */ + readonly startOffsetMs: number; + /** Playable duration of the segment. */ + readonly durationMs: number; + readonly sampleRate: number; + /** + * Every output channel in order, each with the sources heard in it at any + * point in this segment (empty when none were). + */ + readonly channels: readonly AudioChannel[]; + /** Media or container type of the segment bytes, such as `audio/pcm`. */ + readonly mimeType: string; + /** Audio codec, such as `pcm` or `opus`. */ + readonly encoding: string; + /** + * Format details of the resulting file, published with it, such as + * `pcm.sample_format`. Use namespaced keys (`"acme.key_id"`). + */ + readonly attributes?: Readonly>; + /** Private values for later middleware, including binary data. Never published. */ + readonly state?: Readonly>; +} + +type AudioAttributeValue = + | string + | number + | boolean + | null + | readonly AudioAttributeValue[] + | { readonly [key: string]: AudioAttributeValue }; + +/** One output channel: the recorder source keys, as given by the integration, it carries. */ +type AudioChannel = { readonly sources: readonly string[] }; + +export type AudioResult = { segment: Uint8Array; meta: AudioSegmentMeta }; + +/** + * Transforms recorded audio segments, Rack/Koa style. Call `next` at most once + * to run the rest of the stack; it resolves to the final segment and metadata. + * Metadata is frozen: pass or return a new object, such as `{ ...meta }`. + * + * The stack receives PCM bytes. Each layer may pass a different segment type + * down (`Down`), such as codec packets for a container, but the stack must end + * with bytes, and results returned up are always bytes. + * + * Middleware runs once per segment on the main thread, one segment at a time per + * recording, while recordings run concurrently. Keep per-segment state inside + * the call. A throw or rejection omits that segment and stops the recording. + * + * @example + * ```ts + * const logEncoding: AudioMiddleware = async (segment, meta, next) => { + * const result = await next(segment, meta); + * console.log(meta.encoding, "->", result.meta.encoding); + * return result; + * }; + * ``` + */ +export type AudioMiddleware = ( + segment: In, + meta: AudioSegmentMeta, + next: (segment: Down, meta: AudioSegmentMeta) => Promise, +) => AudioResult | Promise; + +/** A stack of middleware whose segment types chain from layer to layer. */ +// eslint-disable-next-line @typescript-eslint/no-explicit-any -- Each layer has its own segment types. +export type AudioMiddlewareStack = readonly AudioMiddleware[]; + +/** + * Runs one segment through the stack and resolves to the bytes and metadata + * that come back out. Rejects if a layer throws or breaks the contract. + */ +export function runMiddleware( + stack: AudioMiddlewareStack, + segment: Uint8Array, + meta: AudioSegmentMeta, +): Promise { + const original = freeze(meta); + const run = async ( + layer: number, + segment: unknown, + meta: unknown, + ): Promise => { + const middleware = stack[layer]; + // The end of the stack returns what reached it, which must be bytes. + if (!middleware) return validate({ segment, meta }, original); + if (!preservesIdentity(meta, original)) + throw new Error("invalid_audio_middleware_result"); + let called = false; + const next = (segment: unknown, meta: AudioSegmentMeta) => { + if (called) throw new Error("next() called multiple times"); + called = true; + const result = run(layer + 1, segment, meta); + // Middleware may leave `next` unawaited; it still sees the rejection if awaited. + result.catch(() => {}); + return result; + }; + return validate(await middleware(segment, freeze(meta), next), original); + }; + return run(0, segment, original); +} + +function freeze(meta: AudioSegmentMeta): AudioSegmentMeta { + if (Object.isFrozen(meta)) return meta; + return Object.freeze({ + ...meta, + channels: Object.freeze( + meta.channels.map((c) => + Object.freeze({ sources: Object.freeze([...c.sources]) }), + ), + ), + ...(meta.attributes && { + attributes: Object.freeze({ ...meta.attributes }), + }), + ...(meta.state && { state: Object.freeze({ ...meta.state }) }), + }); +} + +/** Segment identity and timeline placement survive every transform. */ +function validate(result: unknown, original: AudioSegmentMeta): AudioResult { + const segment = isObject(result) ? result.segment : undefined; + const meta = isObject(result) ? result.meta : undefined; + if (!(segment instanceof Uint8Array) || !preservesIdentity(meta, original)) + throw new Error("invalid_audio_middleware_result"); + return { segment, meta: freeze(meta) }; +} + +function preservesIdentity( + meta: unknown, + original: AudioSegmentMeta, +): meta is AudioSegmentMeta { + return ( + isObject(meta) && + meta.index === original.index && + meta.startOffsetMs === original.startOffsetMs && + Array.isArray(meta.channels) + ); +} diff --git a/js/src/audio-recorder.test.ts b/js/src/audio-recorder.test.ts new file mode 100644 index 000000000..0477c1be4 --- /dev/null +++ b/js/src/audio-recorder.test.ts @@ -0,0 +1,298 @@ +import { expect, test, vi } from "vitest"; +import type { AudioMiddleware, AudioSegmentMeta } from "./audio-middleware"; +import { AudioRecorder } from "./audio-recorder"; + +const mono = (offsetMs: number) => ({ + sampleRate: 1000, + channelCount: 1, + offsetMs, +}); +const ramp = (length: number, from = 1) => + Int16Array.from({ length }, (_, i) => from + i); +const samples = (data: Uint8Array) => + Array.from(new Int16Array(data.slice().buffer)); + +/** A 1 kHz recorder with sources "a" and "b", collecting segments and errors. */ +function recorder( + options: Partial[0]> = {}, +) { + const flushed: { segment: Uint8Array; meta: AudioSegmentMeta }[] = []; + const errors: { meta: AudioSegmentMeta; reason: string }[] = []; + const r = new AudioRecorder({ + sampleRate: 1000, + sources: ["a", "b"], + segmentBytes: 40, // 10 stereo frames: 10 ms. + onFlush(segment, meta) { + flushed.push({ segment, meta }); + }, + onError(meta, reason) { + errors.push({ meta, reason }); + }, + ...options, + }); + return { r, flushed, errors }; +} + +test("without middleware, segments carry the recorded PCM and its format and placement", async () => { + const { r, flushed, errors } = recorder(); + // Leading silence is skipped in whole segments, so 20 ms starts the first one. + expect(r.record("a", ramp(4), mono(20))).toBe(true); + expect(r.record("b", Int16Array.from([-1, -2]), mono(21))).toBe(true); + await r.finish(); + expect(errors).toEqual([]); + expect(flushed).toHaveLength(1); + expect(samples(flushed[0].segment)).toEqual([1, 0, 2, -1, 3, -2, 4, 0]); + expect(flushed[0].meta).toEqual({ + index: 0, + startOffsetMs: 20, + durationMs: 4, + sampleRate: 1000, + channels: [{ sources: ["a"] }, { sources: ["b"] }], + mimeType: "audio/pcm", + encoding: "pcm", + attributes: { + "pcm.sample_format": "s16le", + "pcm.channel_layout": "interleaved", + }, + }); +}); + +test("segments wait for advance() and rotate at the segment size", async () => { + const { r, flushed } = recorder(); + r.record("a", ramp(25), mono(0)); + r.advance(5); + await Promise.resolve(); + expect(flushed).toEqual([]); + r.advance(22); + await vi.waitFor(() => expect(flushed).toHaveLength(2)); + expect( + flushed.map(({ meta }) => [meta.startOffsetMs, meta.durationMs]), + ).toEqual([ + [0, 10], + [10, 10], + ]); + await r.finish(); + expect( + flushed.map(({ meta }) => [ + meta.index, + meta.startOffsetMs, + meta.durationMs, + ]), + ).toEqual([ + [0, 0, 10], + [1, 10, 10], + [2, 20, 5], + ]); +}); + +test("late writes behind the newest audio land in segments that have not been confirmed", async () => { + const { r, flushed } = recorder(); + r.record("a", ramp(20), mono(0)); + r.advance(10); + await vi.waitFor(() => expect(flushed).toHaveLength(1)); + // Output confirmed after input moved on, but still ahead of the confirmed offset. + expect(r.record("b", Int16Array.from([-7, -8]), mono(12))).toBe(true); + await r.finish(); + expect(samples(flushed[1].segment).slice(4, 8)).toEqual([13, -7, 14, -8]); + expect(flushed[1].meta.channels).toEqual([ + { sources: ["a"] }, + { sources: ["b"] }, + ]); + // Every channel is listed; one nobody was heard in has no sources. + expect(flushed[0].meta.channels).toEqual([ + { sources: ["a"] }, + { sources: [] }, + ]); +}); + +test("advance() drains confirmed audio without new writes", async () => { + const { r, flushed } = recorder(); + r.record("a", ramp(30), mono(0)); + r.advance(30); + await vi.waitFor(() => expect(flushed).toHaveLength(3)); + expect(r.active).toBe(true); +}); + +test("buffering stays bounded while advance() stalls", async () => { + const { r, flushed } = recorder({ maxBufferBytes: 40 }); + for (let at = 0; at < 20; at += 5) + expect(r.record("a", ramp(5), mono(at))).toBe(true); + await Promise.resolve(); + expect(flushed).toEqual([]); + expect(r.record("a", ramp(5), mono(20))).toBe(false); + expect(r.reason).toBe("capture_byte_limit"); + await r.finish(); + expect(flushed.map(({ meta }) => meta.durationMs)).toEqual([10, 10]); +}); + +test("rotation and the final partial segment reconstruct both channels exactly", async () => { + const { r, flushed } = recorder(); + const expected: number[] = []; + for (let at = 0; at < 37; at += 3) { + r.record("a", ramp(Math.min(3, 37 - at), at + 1), mono(at)); + r.record( + "b", + ramp(Math.min(3, 37 - at), at + 1).map((v) => -v), + mono(at), + ); + r.advance(at + 3); + } + for (let i = 0; i < 37; i++) expected.push(i + 1, -(i + 1)); + await r.finish(); + expect(flushed.flatMap(({ segment }) => samples(segment))).toEqual(expected); + expect(flushed.map(({ meta }) => meta.durationMs)).toEqual([10, 10, 10, 7]); + // Segments tile the timeline without gaps or overlaps. + for (const [i, { meta }] of flushed.entries()) + if (i) + expect(meta.startOffsetMs).toBe( + flushed[i - 1].meta.startOffsetMs + flushed[i - 1].meta.durationMs, + ); +}); + +test("middleware output is flushed; a failing segment is omitted, stops capture and keeps earlier audio", async () => { + let calls = 0; + const failSecond: AudioMiddleware = (segment, meta, next) => { + if (++calls === 2) throw new Error("boom"); + return next(segment, { ...meta, encoding: "tagged" }); + }; + const { r, flushed, errors } = recorder({ middleware: [failSecond] }); + r.record("a", ramp(30), mono(0)); + r.advance(30); + await vi.waitFor(() => expect(errors).toHaveLength(1)); + expect(errors[0]).toMatchObject({ + meta: { index: 1, startOffsetMs: 10 }, + reason: "recording_transform_failed", + }); + expect(r.reason).toBe("recording_transform_failed"); + expect(r.record("a", ramp(1), mono(40))).toBe(false); + await r.finish(); + // Accepted audio after the failure still drains on finish. + expect(flushed.map(({ meta }) => [meta.index, meta.encoding])).toEqual([ + [0, "tagged"], + [2, "tagged"], + ]); +}); + +test("a rejected flush is reported, stops capture and never rejects finish", async () => { + const { r, errors } = recorder({ + onFlush: async () => { + throw new Error("offline"); + }, + }); + r.record("a", Int16Array.from([1]), mono(0)); + await expect(r.finish()).resolves.toBeUndefined(); + expect(errors.map(({ reason }) => reason)).toEqual([ + "recording_export_failed", + ]); + expect(r.reason).toBe("recording_export_failed"); +}); + +test("one segment is in flight; slow work bounds the backlog and keeps the accepted prefix", async () => { + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const slow: AudioMiddleware = async (segment, meta, next) => { + await gate; + return next(segment, meta); + }; + // Room for 20 buffered samples; the in-flight segment sits outside the budget. + const { r, flushed } = recorder({ middleware: [slow], maxBufferBytes: 40 }); + expect(r.record("a", ramp(10), mono(0))).toBe(true); + r.advance(10); + for (let at = 10; at < 30; at += 10) { + expect(r.record("a", ramp(10), mono(at))).toBe(true); + r.advance(at + 10); + } + expect(r.record("a", ramp(10), mono(30))).toBe(false); + expect(r.reason).toBe("capture_byte_limit"); + release(); + await r.finish(); + expect(flushed.map(({ meta }) => meta.startOffsetMs)).toEqual([0, 10, 20]); +}); + +test("finish is idempotent and waits for flushes to settle", async () => { + let settled = false; + const { r } = recorder({ + onFlush: async () => { + await new Promise((resolve) => setTimeout(resolve, 5)); + settled = true; + }, + }); + r.record("a", Int16Array.from([1]), mono(0)); + const finishing = r.finish(); + expect(r.finish()).toBe(finishing); + expect(r.closed).toBe(true); + await finishing; + expect(settled).toBe(true); + expect(r.record("a", Int16Array.from([1]), mono(1))).toBe(false); +}); + +test("staged reservations share the buffer budget until released", async () => { + const { r, flushed } = recorder({ maxBufferBytes: 16, segmentBytes: 16 }); + expect(r.reserve(12)).toBe(true); + expect(r.record("a", ramp(2), mono(0))).toBe(true); + expect(r.record("a", ramp(1), mono(2))).toBe(false); + expect(r.reason).toBe("capture_byte_limit"); + await r.finish(); + expect(flushed.flatMap(({ segment }) => samples(segment))).toEqual([ + 1, 0, 2, 0, + ]); + + const released = recorder({ maxBufferBytes: 16, segmentBytes: 16 }); + expect(released.r.reserve(12)).toBe(true); + released.r.release(12); + expect(released.r.record("a", ramp(8), mono(0))).toBe(true); + expect(released.r.reserve(1)).toBe(false); +}); + +test("the duration limit and invalid audio stop capture with a reason", async () => { + const limited = recorder({ maxDurationMs: 10 }); + expect(limited.r.record("a", ramp(8), mono(0))).toBe(true); + expect(limited.r.record("a", ramp(4), mono(8))).toBe(false); + expect(limited.r.reason).toBe("duration_limit"); + await limited.r.finish(); + expect(limited.flushed[0].meta.durationMs).toBe(8); + + const invalid = recorder(); + expect(invalid.r.record("a", ramp(3), { ...mono(0), channelCount: 2 })).toBe( + false, + ); + expect(invalid.r.reason).toBe("incomplete_pcm_frame"); + invalid.r.stop("later"); + expect(invalid.r.reason).toBe("incomplete_pcm_frame"); + + const unknown = recorder(); + expect(unknown.r.record("c", ramp(1), mono(0))).toBe(false); + expect(unknown.r.reason).toBe("unknown_audio_source"); +}); + +test("concurrent recorders with shared middleware keep their segments separate", async () => { + const gates: (() => void)[] = []; + const tag: AudioMiddleware = async (segment, meta, next) => { + await new Promise((resolve) => gates.push(resolve)); + return next(segment, { ...meta, encoding: `tagged-${segment[0]}` }); + }; + const a = recorder({ middleware: [tag] }); + const b = recorder({ middleware: [tag] }); + a.r.record("a", ramp(15, 1), mono(0)); + b.r.record("a", ramp(15, 50), mono(0)); + const finishing = Promise.all([a.r.finish(), b.r.finish()]); + // Release in reverse arrival order; each recorder still flushes in order. + while (a.flushed.length + b.flushed.length < 4) { + gates + .splice(0) + .reverse() + .forEach((release) => release()); + await new Promise((resolve) => setTimeout(resolve, 0)); + } + await finishing; + expect(a.flushed.map(({ meta }) => [meta.index, meta.encoding])).toEqual([ + [0, "tagged-1"], + [1, "tagged-11"], + ]); + expect(b.flushed.map(({ meta }) => [meta.index, meta.encoding])).toEqual([ + [0, "tagged-50"], + [1, "tagged-60"], + ]); +}); diff --git a/js/src/audio-recorder.ts b/js/src/audio-recorder.ts new file mode 100644 index 000000000..8bca00922 --- /dev/null +++ b/js/src/audio-recorder.ts @@ -0,0 +1,229 @@ +import { debugLogger } from "./debug-logger"; +import { + runMiddleware, + type AudioMiddlewareStack, + type AudioResult, + type AudioSegmentMeta, +} from "./audio-middleware"; +import { PcmBuffer, pcmBytesToMs } from "./pcm-buffer"; + +type AudioRecorderOptions = { + sampleRate: number; + /** Source keys, one per output channel in order. */ + sources: readonly string[]; + /** Target size of each segment entering the middleware. */ + segmentBytes: number; + /** Bound on buffered and reserved audio; one in-flight segment is extra. */ + maxBufferBytes?: number; + maxDurationMs?: number; + middleware?: AudioMiddlewareStack; + /** Receives each transformed segment; the recorder waits for it to settle. */ + onFlush(segment: Uint8Array, meta: AudioSegmentMeta): void | Promise; + /** Receives the original metadata of a segment that failed, and the stop reason. */ + onError(meta: AudioSegmentMeta, reason: string, error: unknown): void; +}; + +const PCM_ATTRIBUTES = { + "pcm.sample_format": "s16le", + "pcm.channel_layout": "interleaved", +}; + +/** + * Assembles PCM on a channel timeline, cuts bounded segments behind a confirmed + * offset, and passes them through middleware to `onFlush`, one at a time. + */ +export class AudioRecorder { + private buffer: PcmBuffer; + private confirmedThroughMs = 0; + private reservedBytes = 0; + private index = 0; + private job?: Promise; + private finishing?: Promise; + private warned = false; + private stopReason?: string; + private isClosed = false; + + constructor(private options: AudioRecorderOptions) { + const maxBufferBytes = + options.maxBufferBytes ?? + Math.min( + Number.MAX_SAFE_INTEGER, + Math.max(32 * 1024 * 1024, 2 * options.segmentBytes), + ); + if (maxBufferBytes < options.segmentBytes) + throw new Error("invalid_buffer_limit"); + this.buffer = new PcmBuffer({ + sampleRate: options.sampleRate, + channels: options.sources.length, + maxBufferBytes, + }); + } + + get reason(): string | undefined { + return this.stopReason; + } + get closed(): boolean { + return this.isClosed; + } + /** Whether new audio is accepted. */ + get active(): boolean { + return !this.isClosed && !this.stopReason; + } + /** Stops accepting audio. Accepted audio still flushes on finish. */ + stop(reason: string): void { + this.stopReason ??= reason; + } + + /** + * Writes positioned PCM for a source, stopping on any limit or invalid audio. + * `channelCount` is the number of interleaved channels in `samples`; they are + * downmixed onto the source's channel. + */ + record( + source: string, + samples: Int16Array, + format: { sampleRate: number; channelCount: number; offsetMs: number }, + ): boolean { + if (!this.active) return false; + const channel = this.options.sources.indexOf(source); + if (channel < 0) { + this.stop("unknown_audio_source"); + return false; + } + const end = + format.offsetMs + + pcmBytesToMs(samples.byteLength, format.sampleRate, format.channelCount); + if (end > (this.options.maxDurationMs ?? Infinity)) { + this.stop("duration_limit"); + return false; + } + const bytes = + Math.round( + ((samples.length / format.channelCount) * this.buffer.sampleRate) / + format.sampleRate, + ) * 2; + if (!this.fits(bytes)) return false; + try { + this.buffer.write(samples, { + sampleRate: format.sampleRate, + channels: format.channelCount, + channel, + offsetMs: format.offsetMs, + }); + } catch (error) { + this.stop(error instanceof Error ? error.message : "capture_failed"); + return false; + } + return true; + } + + /** Reserves budget for audio held before it can be recorded. */ + reserve(bytes: number): boolean { + if (!this.active || !this.fits(bytes)) return false; + this.reservedBytes += bytes; + return true; + } + release(bytes: number): void { + this.reservedBytes -= bytes; + } + + /** No audio will be recorded before `throughMs`, so earlier audio may flush. */ + advance(throughMs: number): void { + this.confirmedThroughMs = Math.max(this.confirmedThroughMs, throughMs); + this.pump(); + } + + /** Stops capture, flushes the remainder, and resolves once every flush settles. */ + finish(): Promise { + return (this.finishing ??= (async () => { + this.isClosed = true; + this.pump(); + while (this.job) await this.job; + })()); + } + + private fits(bytes: number): boolean { + if ( + this.buffer.bufferedByteLength + this.reservedBytes + bytes <= + this.buffer.maxBufferBytes + ) + return true; + this.stop("capture_byte_limit"); + return false; + } + + private pump(): void { + if ( + this.job || + !this.buffer.bufferedByteLength || + (this.stopReason && !this.isClosed) + ) + return; + let cut: ReturnType; + try { + cut = this.buffer.flush({ + maxBytes: this.options.segmentBytes, + endOffsetMs: this.isClosed + ? undefined + : Math.min(this.confirmedThroughMs, this.buffer.endOffsetMs), + partial: this.isClosed, + }); + } catch (error) { + this.stop( + error instanceof Error ? error.message : "recording_export_failed", + ); + return; + } + if (!cut) return; + const meta: AudioSegmentMeta = { + index: this.index++, + startOffsetMs: cut.startOffsetMs, + durationMs: cut.endOffsetMs - cut.startOffsetMs, + sampleRate: this.buffer.sampleRate, + // One channel per source, listing its source when it was heard. + channels: this.options.sources.map((source, channel) => ({ + sources: cut.channels.includes(channel) ? [source] : [], + })), + mimeType: "audio/pcm", + encoding: "pcm", + attributes: PCM_ATTRIBUTES, + }; + this.job = this.process(cut.data, meta).finally(() => { + this.job = undefined; + this.pump(); + }); + } + + private async process(segment: Uint8Array, meta: AudioSegmentMeta) { + let result: AudioResult; + try { + result = await runMiddleware( + this.options.middleware ?? [], + segment, + meta, + ); + } catch (error) { + if (!this.warned) + debugLogger.notice.warn( + "Audio middleware failed; the recording omits this segment and stops.", + error, + ); + this.warned = true; + return this.fail("recording_transform_failed", error, meta); + } + try { + await this.options.onFlush(result.segment, result.meta); + } catch (error) { + this.fail("recording_export_failed", error, meta); + } + } + + private fail(reason: string, error: unknown, meta: AudioSegmentMeta) { + this.stop(reason); + try { + this.options.onError(meta, reason, error); + } catch { + // Reporting must not break draining. + } + } +} diff --git a/js/src/instrumentation/plugins/livekit-plugin.test.ts b/js/src/instrumentation/plugins/livekit-plugin.test.ts index d05361e97..68cb3b8f5 100644 --- a/js/src/instrumentation/plugins/livekit-plugin.test.ts +++ b/js/src/instrumentation/plugins/livekit-plugin.test.ts @@ -17,6 +17,7 @@ import { initLogger, } from "../../logger"; import type { LiveKitSpan, SpanTime } from "../../vendor-sdk-types/livekit"; +import type { AudioMiddleware } from "../../audio-middleware"; import { DEFAULT_STREAMING_ATTACHMENT_BUFFER_BYTES } from "../config"; import { channel, defineInterceptor } from "../core/interceptor-definitions"; import { traceAsyncCall } from "../core/tracing"; @@ -1005,6 +1006,154 @@ describe("recording", () => { expect(recordings).toBeDefined(); }); + /** Replaces the default tracing with one that runs the given middleware. */ + async function withMiddleware( + middleware: AudioMiddleware, + chunkBytes = DEFAULT_STREAMING_ATTACHMENT_BUFFER_BYTES, + ) { + await tracing.shutdown(); + tracing = new LiveKitTracing(chunkBytes, () => [middleware]); + } + /** One agent frame played for 20 ms starting 10 ms into the call. */ + async function playOneFrame(sink: EventEmitter & any, origin: number) { + await sink.captureFrame(frame()); + sink.emit("playbackStarted", { createdAt: origin + 10 }); + sink.emit("playbackFinished", { + playbackPosition: 0.02, + interrupted: false, + }); + } + + test("disabled audio capture never runs middleware", async () => { + vi.stubEnv("BRAINTRUST_CAPTURE_USER_AUDIO_ATTACHMENTS", ""); + vi.stubEnv("BRAINTRUST_CAPTURE_AGENT_AUDIO_ATTACHMENTS", ""); + const middleware = vi.fn(); + await withMiddleware(middleware); + const { sink, origin, close } = session(); + await playOneFrame(sink, origin); + expect((await close()).recordings).toBeUndefined(); + expect(middleware).not.toHaveBeenCalled(); + }); + + test("middleware output is uploaded with its format; selections are unchanged", async () => { + await withMiddleware(async (segment, meta, next) => { + const framed = new Uint8Array(segment.length + 4); + framed.set([0x54, 0x45, 0x53, 0x54]); + framed.set(segment, 4); + return next(framed, { + ...meta, + mimeType: "audio/x-test", + encoding: "test", + attributes: { "acme.framing": "test" }, + state: { "acme.header": framed.subarray(0, 4) }, + }); + }); + const { sink, origin, close } = session(); + await playOneFrame(sink, origin); + const { recordings, selections, named } = await close(); + expect(recordings).toMatchObject([ + { + id: "call-0000", + state: "ready", + mime_type: "audio/x-test", + duration_ms: 30, // Includes the 10 ms before playout started. + sources: [{ channel_index: 1 }], + }, + ]); + // Only raw PCM publishes a sample format. + expect(recordings[0]).not.toHaveProperty("sample_format"); + // Pipeline attributes and state are not published in the recording descriptor. + expect(JSON.stringify(recordings)).not.toContain("acme."); + expect(selections("agent_turn")).toEqual([["call-0000", 10, 30]]); + expect([...uploads[0].slice(0, 4)]).toEqual([0x54, 0x45, 0x53, 0x54]); + expect(channelSamples(uploads[0].slice(4), 1)[300]).toBe(1234); + const audio = named("session").find((row) => row.input?.audio)?.input.audio[ + "call-0000" + ]; + expect(audio).toMatchObject({ + filename: "call-0000.x-test", + content_type: "audio/x-test", + }); + }); + + test("failing middleware omits its file without changing sink I/O", async () => { + vi.spyOn(console, "warn").mockImplementation(() => {}); + await withMiddleware(() => { + throw new Error("encoder unavailable"); + }); + const { sink, origin, close } = session(); + await expect(playOneFrame(sink, origin)).resolves.toBeUndefined(); + const { recordings, selections } = await close(); + expect(recordings).toEqual([ + expect.objectContaining({ + id: "call-0000", + state: "omitted", + reason: "recording_transform_failed", + gaps: [ + { + start_offset_ms: 0, + end_offset_ms: 30, + reason: "recording_transform_failed", + }, + ], + }), + { + id: "call", + recording_group_id: "call", + state: "omitted", + reason: "recording_transform_failed", + truncated: true, + }, + ]); + expect(selections("agent_turn")).toEqual([]); + expect(uploads).toEqual([]); + }); + + test("slow middleware never delays playback; close waits for it", async () => { + let entered = 0; + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + // 100 ms segments of 24 kHz stereo s16le. + await withMiddleware(async (segment, meta, next) => { + entered++; + await gate; + return next(segment, meta); + }, 9600); + const { sink, origin, close } = session(); + const play = async (startedAt: number) => { + await sink.captureFrame(frame(2400)); + sink.emit("playbackStarted", { createdAt: startedAt }); + sink.emit("playbackFinished", { + playbackPosition: 0.1, + interrupted: false, + }); + }; + await play(origin); + now = origin + 2000; // The first 100 ms are now safe to cut. + await play(origin + 1500); + expect(entered).toBe(1); + // Later playback completes while that segment is held in middleware. + await play(origin + 1700); + expect(uploads).toEqual([]); + const closing = close(); + release(); + const { recordings } = await closing; + // The empty interval between playouts is skipped, not uploaded as silence. + expect( + recordings.map((r: any) => [ + r.state, + r.timeline.recording_start_offset_ms, + r.duration_ms, + ]), + ).toEqual([ + ["ready", 0, 100], + ["ready", 1500, 100], + ["ready", 1700, 100], + ]); + }); + test("interruption records only confirmed progress ranges, excluding the queued tail and pauses", async () => { const { sink, origin, close } = session(); await sink.captureFrame(frame(2400)); // 100 ms generated @@ -1282,12 +1431,16 @@ describe("recording", () => { }); describe("recording files", () => { - const packet = (at: number, length = 24000, channel = 0) => ({ + const packet = ( + at: number, + length = 24000, + source: "session_input" | "local_audio_output" = "session_input", + ) => ({ pcm: Int16Array.from({ length }, (_, i) => ((i * 31) % 30000) - 15000), at, rate: 24000, channels: 1, - channel, + source, }); /** A recording whose wall-clock hold no longer limits uploads. */ function recording(chunkBytes = DEFAULT_STREAMING_ATTACHMENT_BUFFER_BYTES) { @@ -1330,11 +1483,14 @@ describe("recording files", () => { return { upload_status: "done" }; }); const { r, log, files, selections } = recording(96000); - r.select("turn", 0, 1000, 0); + r.select("turn", 0, 1000, "session_input"); r.publish(["turn"]); r.record(packet(0)); r.advance(); - expect(files()).toEqual([["call-0000", "pending", 0, undefined]]); + // Pending once the segment leaves the (empty) middleware stack. + await vi.waitFor(() => + expect(files()).toEqual([["call-0000", "pending", 0, undefined]]), + ); expect(log.mock.calls.some(([event]) => event.input)).toBe(false); expect(selections()).toEqual([]); complete(); @@ -1357,7 +1513,13 @@ describe("recording files", () => { (_, i) => (((i + at) * (channel ? 31 : 17)) % 30001) - 15000, ); expect( - r.record({ pcm, at: at / 24, rate: 24000, channels: 1, channel }), + r.record({ + pcm, + at: at / 24, + rate: 24000, + channels: 1, + source: channel ? "local_audio_output" : "session_input", + }), ).toBe(true); for (let i = 0; i < pcm.length; i++) expected.writeInt16LE(pcm[i], ((at + i) * 2 + channel) * 2); @@ -1409,7 +1571,7 @@ describe("recording files", () => { error_message: "offline", }); const { r, files, selections } = recording(96000); - r.select("turn", 0, 2000, 0); + r.select("turn", 0, 2000, "session_input"); for (let i = 0; i < 2; i++) { r.record(packet(i * 1000)); r.advance(); @@ -1429,10 +1591,10 @@ describe("recording files", () => { test("late selections split across ready files and merge adjacent intervals", async () => { const { r, selections } = recording(96000); - r.record(packet(0, 48000)); + r.record(packet(0, 48000, "local_audio_output")); await r.finish(); - r.select("turn", 500, 1000, 1); - r.select("turn", 1000, 1500, 1); + r.select("turn", 500, 1000, "local_audio_output"); + r.select("turn", 1000, 1500, "local_audio_output"); r.publish(["turn"]); expect(selections()).toEqual([ ["call-0000", 500, 1000], diff --git a/js/src/instrumentation/plugins/livekit-plugin.ts b/js/src/instrumentation/plugins/livekit-plugin.ts index 6218632f9..63772704c 100644 --- a/js/src/instrumentation/plugins/livekit-plugin.ts +++ b/js/src/instrumentation/plugins/livekit-plugin.ts @@ -22,7 +22,7 @@ import { runInstrumentation } from "../core/observe-result"; import { livekitChannels } from "./livekit-interceptors"; import { v4 as uuidv4 } from "uuid"; import { runWithAutoInstrumentationSuppressed } from "../auto-instrumentation-suppression"; -import { startSpan, withCurrent, type Span } from "../../logger"; +import { Attachment, startSpan, withCurrent, type Span } from "../../logger"; import { withSpanInstrumentationName, INSTRUMENTATION_NAMES, @@ -32,12 +32,12 @@ import { zodToJsonSchema } from "../../zod/utils"; import { debugLogger } from "../../debug-logger"; import { isEnvFlagEnabled } from "../../wrappers/attachment-utils"; import { DEFAULT_STREAMING_ATTACHMENT_BUFFER_BYTES } from "../config"; -import { - PcmAudio, - pcmBytesToMs, - msToSamples, - samplesToMs, -} from "../../pcm-audio"; +import { pcmBytesToMs, msToSamples, samplesToMs } from "../../pcm-buffer"; +import { AudioRecorder } from "../../audio-recorder"; +import type { + AudioMiddlewareStack, + AudioSegmentMeta, +} from "../../audio-middleware"; // Bounds on per-session observation state, independent of the attachment threshold. const MAX_ROWS = 8192; @@ -237,7 +237,10 @@ class LiveKitTracing { // Callbacks already handed to LiveKit can outlive interceptor unsubscription. private disabled = false; - constructor(private streamingAttachmentBufferBytes: number) {} + constructor( + private streamingAttachmentBufferBytes: number, + private audioMiddleware?: () => AudioMiddlewareStack, + ) {} private rowOf(span: LiveKitSpan | undefined): Row | undefined { const row = span && observedRows.get(span); @@ -408,6 +411,7 @@ class LiveKitTracing { this.streamingAttachmentBufferBytes, (id) => rows.get(id), captures, + this.audioMiddleware, ) : undefined, closed: false, @@ -1128,21 +1132,24 @@ class SpeechTracker { } } +/** Recorder source keys, published as each recording channel's boundary. */ +type Source = "session_input" | "local_audio_output"; type Packet = { pcm: Int16Array; rate: number; channels: number; - channel: number; + source: Source; /** Session-clock offset in milliseconds; relative to its run while staged. */ at: number; }; -type Interval = { start: number; end: number; channel: number }; +type Interval = { start: number; end: number; source: Source }; type AudioFile = { id: string; start: number; end: number; - channels: number[]; - metadata: NonNullable>["metadata"]; + /** Each source heard in the file and its channel. */ + sources: { channel: number; name: string }[]; + metadata?: Record; state: "pending" | "ready" | "omitted"; reason?: string; }; @@ -1150,51 +1157,67 @@ type AudioFile = { const inputs = new WeakMap(); const sinks = new WeakMap(); -/** LiveKit's call clock, upload queue, and speech-to-file references. PcmAudio owns the bytes. */ +/** LiveKit's call clock, uploads, and speech-to-file references. AudioRecorder owns the audio. */ class Recording { readonly origin = Date.now(); /** Session-clock playout that each pending output run may still add audio before. */ readonly holds = new Map(); - private pcm: PcmAudio; + private recorder: AudioRecorder; private files: AudioFile[] = []; - private formats = [ - new Map(), - new Map(), - ]; + private formats: Record< + Source, + Map + > = { session_input: new Map(), local_audio_output: new Map() }; private selections = new Map(); private playbackEvents = new Map(); - private confirmedThroughMs = 0; - private stagedBytes = 0; private inputStream?: object; private inputStart?: number; private inputEnd?: number; - private job?: Promise; private finishing?: Promise; - reason?: string; - closed = false; constructor( private span: Span, - private chunkBytes: number, + chunkBytes: number, private resolve: (owner: string) => Row | undefined, /** Which directions this recording captures: caller input, agent output. */ readonly captures = { input: true, output: true }, + middleware?: () => AudioMiddlewareStack, ) { - this.pcm = new PcmAudio({ + // Each recording builds its own stack, so plugin state is never shared. + let stack: AudioMiddlewareStack = []; + let failed = false; + try { + stack = middleware?.() ?? []; + } catch (error) { + failed = true; + debugLogger.notice.warn( + "LiveKit audio middleware failed to start.", + error, + ); + } + this.recorder = new AudioRecorder({ sampleRate: SAMPLE_RATE, - channels: 2, - maxBufferBytes: Math.min( - Number.MAX_SAFE_INTEGER, - Math.max(32 * 1024 * 1024, 2 * chunkBytes), - ), + sources: ["session_input", "local_audio_output"], + segmentBytes: chunkBytes, + maxDurationMs: MAX_RECORDING_MS, + middleware: stack, + onFlush: (segment, meta) => this.upload(segment, meta), + onError: (meta, reason) => this.omit(meta, reason), }); + if (failed) this.stop("recording_transform_failed"); } get active(): boolean { - return !this.closed && !this.reason; + return this.recorder.active; + } + get closed(): boolean { + return this.recorder.closed; + } + get reason(): string | undefined { + return this.recorder.reason; } stop(reason: string): void { - this.reason ??= reason; + this.recorder.stop(reason); } /** Validates a LiveKit frame, returning its samples while capture continues. */ @@ -1215,39 +1238,15 @@ class Recording { } record(packet: Packet): boolean { - if (!this.active) return false; - const durationMs = pcmBytesToMs( - packet.pcm.byteLength, - packet.rate, - packet.channels, - ); - if (packet.at + durationMs > MAX_RECORDING_MS) { - this.stop("duration_limit"); - return false; - } - const bytes = - Math.round( - ((packet.pcm.length / packet.channels) * SAMPLE_RATE) / packet.rate, - ) * 2; if ( - this.pcm.bufferedByteLength + this.stagedBytes + bytes > - this.pcm.maxBufferBytes - ) { - this.stop("capture_byte_limit"); - return false; - } - try { - this.pcm.update(packet.pcm, { + !this.recorder.record(packet.source, packet.pcm, { sampleRate: packet.rate, - channels: packet.channels, - channel: packet.channel, + channelCount: packet.channels, offsetMs: packet.at, - }); - } catch (error) { - this.stop(error instanceof Error ? error.message : "capture_failed"); + }) + ) return false; - } - this.formats[packet.channel].set(`${packet.rate}:${packet.channels}`, { + this.formats[packet.source].set(`${packet.rate}:${packet.channels}`, { sample_rate_hz: packet.rate, channel_count: packet.channels, }); @@ -1257,20 +1256,12 @@ class Recording { /** Copies an output frame until playout confirms or discards it. */ stage(frame: AudioFrame): Int16Array | undefined { const data = this.samples(frame); - if (!data) return; - if ( - this.stagedBytes + this.pcm.bufferedByteLength + data.byteLength > - this.pcm.maxBufferBytes - ) { - this.stop("capture_byte_limit"); - return; - } - this.stagedBytes += data.byteLength; + if (!data || !this.recorder.reserve(data.byteLength)) return; // LiveKit may reuse the frame while captureFrame is awaiting the sink. return data.slice(); } unstage(pcm: Int16Array): void { - this.stagedBytes -= pcm.byteLength; + this.recorder.release(pcm.byteLength); } attachInput(stream: ReadableStream): void { @@ -1308,7 +1299,7 @@ class Recording { pcm, rate: frame.sampleRate, channels: frame.channels, - channel: 0, + source: "session_input", at, }) ) @@ -1330,18 +1321,18 @@ class Recording { owner, Math.max(this.inputStart, fromUnixMs - this.origin), Math.min(this.inputEnd!, toUnixMs - this.origin), - 0, + "session_input", ); this.publish([owner]); } - select(owner: string, start: number, end: number, channel: number): void { + select(owner: string, start: number, end: number, source: Source): void { if (!(end > start)) return; const list = this.selections.get(owner) ?? []; const previous = list.at(-1); if ( previous && - previous.channel === channel && + previous.source === source && start >= previous.start && start <= previous.end + samplesToMs(1) ) { @@ -1352,7 +1343,7 @@ class Recording { this.stop("selection_limit"); return; } - list.push({ start, end, channel }); + list.push({ start, end, source }); this.selections.set(owner, list); } @@ -1365,89 +1356,90 @@ class Recording { /** Only cut where neither input nor a queued playback run can still add audio. */ advance(): void { const wall = Date.now() - this.origin - 1000; - this.confirmedThroughMs = Math.max( - this.confirmedThroughMs, + this.recorder.advance( Math.min( wall, this.inputStream ? (this.inputEnd ?? 0) : wall, ...this.holds.values(), ), ); - this.upload(); } finish(): Promise { - return (this.finishing ??= (async () => { - this.closed = true; - this.upload(); - while (this.job) await this.job; - this.publishFiles(); - })()); + return (this.finishing ??= this.recorder + .finish() + .then(() => this.publishFiles())); } - private upload(): void { - if ( - this.job || - !this.pcm.bufferedByteLength || - (this.reason && !this.closed) - ) - return; - const id = `call-${String(this.files.length).padStart(4, "0")}`; - let result: ReturnType; - try { - result = this.pcm.digest({ - maxBytes: this.chunkBytes, - endOffsetMs: this.closed - ? undefined - : Math.min(this.confirmedThroughMs, this.pcm.endOffsetMs), - flush: this.closed, - filename: `${id}.pcm`, - }); - } catch (error) { - this.stop( - error instanceof Error ? error.message : "recording_export_failed", - ); - return; - } - if (!result) return; - const { attachment } = result; - const file: AudioFile = { - id, - start: result.startOffsetMs, - end: result.endOffsetMs, - channels: result.channels, - metadata: result.metadata, - state: "pending", + /** One ordinary SDK attachment upload per segment; capture continues meanwhile. */ + private async upload( + segment: Uint8Array, + meta: AudioSegmentMeta, + ): Promise { + const file = this.file(meta); + file.metadata = { + mime_type: meta.mimeType, + sample_rate_hz: meta.sampleRate, + channel_count: meta.channels.length, + // Headerless PCM needs its sample format published; files such as Ogg carry their own. + ...(meta.mimeType === "audio/pcm" && { + sample_format: meta.attributes?.["pcm.sample_format"], + }), + duration_ms: meta.durationMs, }; - this.files.push(file); this.publishFiles(); - // One ordinary SDK attachment upload at a time; capture continues into the next digest. - this.job = (async () => { - try { - const status = await attachment.upload(); - file.state = status.upload_status === "done" ? "ready" : "omitted"; - } catch { - file.state = "omitted"; - } - if (file.state === "ready") { - this.span.log({ input: { audio: { [id]: attachment.reference } } }); - } else { - file.reason = "recording_export_failed"; - this.stop(file.reason); - } - this.publishFiles(); - if (file.state === "ready") - this.publish( - [...this.selections] - .filter(([, list]) => - list.some((s) => s.start < file.end && s.end > file.start), - ) - .map(([owner]) => owner), - ); - })().finally(() => { - this.job = undefined; - this.upload(); + const attachment = new Attachment({ + data: segment, + filename: `${file.id}.${fileExtension(meta.mimeType)}`, + contentType: meta.mimeType, }); + try { + const status = await attachment.upload(); + file.state = status.upload_status === "done" ? "ready" : "omitted"; + } catch { + file.state = "omitted"; + } + if (file.state === "ready") { + this.span.log({ input: { audio: { [file.id]: attachment.reference } } }); + } else { + file.reason = "recording_export_failed"; + this.stop(file.reason); + } + this.publishFiles(); + if (file.state === "ready") + this.publish( + [...this.selections] + .filter(([, list]) => + list.some((s) => s.start < file.end && s.end > file.start), + ) + .map(([owner]) => owner), + ); + } + + private omit(meta: AudioSegmentMeta, reason: string): void { + const file = this.file(meta); + file.state = "omitted"; + file.reason = reason; + this.publishFiles(); + } + + /** The file for a segment, keyed by its position in the recording. */ + private file(meta: AudioSegmentMeta): AudioFile { + const id = `call-${String(meta.index).padStart(4, "0")}`; + let file = this.files.find((f) => f.id === id); + if (!file) { + file = { + id, + start: meta.startOffsetMs, + end: meta.startOffsetMs + meta.durationMs, + sources: meta.channels.flatMap((c, channel) => + c.sources.map((name) => ({ channel, name })), + ), + state: "pending", + }; + this.files.push(file); + } + return file; } private publishFiles(): void { @@ -1465,9 +1457,9 @@ class Recording { ...f.metadata, } : {}), - sources: f.channels.map((channel_index) => ({ - channel_index, - boundary: channel_index === 0 ? "session_input" : "local_audio_output", + sources: f.sources.map((s) => ({ + channel_index: s.channel, + boundary: s.name, })), timeline: { origin_unix_ms: this.origin, @@ -1504,8 +1496,12 @@ class Recording { this.span.log({ metadata: { "audio.recordings": recordings, - "contrib.livekit.audio.input_formats": [...this.formats[0].values()], - "contrib.livekit.audio.output_formats": [...this.formats[1].values()], + "contrib.livekit.audio.input_formats": [ + ...this.formats.session_input.values(), + ], + "contrib.livekit.audio.output_formats": [ + ...this.formats.local_audio_output.values(), + ], }, }); } @@ -1522,18 +1518,26 @@ class Recording { row.span.log({ metadata: { "audio.selections": ready.flatMap((f) => - intervals - .map((s) => ({ + intervals.flatMap((s) => { + // The file describes which channel holds each source, if any. + const channel = f.sources.find( + (x) => x.name === s.source, + )?.channel; + const selection = { recording_span_id: this.span.spanId, recording_id: f.id, - channel_index: s.channel, + channel_index: channel, start_offset_ms: Math.max(0, toSample(s.start - f.start)), end_offset_ms: Math.min( f.end - f.start, toSample(s.end - f.start), ), - })) - .filter((s) => s.end_offset_ms > s.start_offset_ms), + }; + return channel !== undefined && + selection.end_offset_ms > selection.start_offset_ms + ? [selection] + : []; + }), ), ...(events ? { "contrib.livekit.playback_events": events } : {}), }, @@ -1542,6 +1546,10 @@ class Recording { } } +function fileExtension(mimeType: string): string { + return mimeType.split(";")[0].split("/")[1]?.trim() || "bin"; +} + // Tap the reader already requested by LiveKit. No tee, prefetch, or extra consumer. // Keeping the original stream lets LiveKit release/reacquire its lock on handoff. function tapInput(stream: any) { @@ -1688,7 +1696,7 @@ class Playback { pcm, rate: frame.sampleRate, channels: frame.channels, - channel: 1, + source: "local_audio_output", at: run.duration, }); run.duration += pcmBytesToMs( @@ -1783,7 +1791,12 @@ class Playback { run.speaking ??= observedId(this.session.agentSpeakingSpan); if (run.speaking) for (const played of run.unclaimed.splice(0)) - recording.select(run.speaking, played.start, played.end, 1); + recording.select( + run.speaking, + played.start, + played.end, + "local_audio_output", + ); const ranges = run.ranges.splice(0); if (finished && !run.hadProgress) { // Sinks without progress events confirm one contiguous run from its start. @@ -1830,9 +1843,20 @@ class Playback { start: at, end: at + pcmBytesToMs(pcm.byteLength, frame.rate, frame.channels), }; - if (run.owner) recording.select(run.owner, played.start, played.end, 1); + if (run.owner) + recording.select( + run.owner, + played.start, + played.end, + "local_audio_output", + ); if (run.speaking) - recording.select(run.speaking, played.start, played.end, 1); + recording.select( + run.speaking, + played.start, + played.end, + "local_audio_output", + ); else if (run.unclaimed.length < MAX_RANGES) run.unclaimed.push(played); else recording.stop("selection_limit"); } diff --git a/js/src/pcm-audio.test.ts b/js/src/pcm-audio.test.ts deleted file mode 100644 index 891a3bb0e..000000000 --- a/js/src/pcm-audio.test.ts +++ /dev/null @@ -1,176 +0,0 @@ -import { expect, test } from "vitest"; -import { PcmAudio } from "./pcm-audio"; - -const MAX = Number.MAX_SAFE_INTEGER; -// Without a chunk-sized limit, flush consumes everything through the cut. -const all = { maxBytes: MAX, flush: true }; -async function bytes(result: ReturnType) { - return Buffer.from(await (await result!.attachment.data()).arrayBuffer()); -} -const samples = (data: Buffer) => - Array.from({ length: data.length / 2 }, (_, i) => data.readInt16LE(i * 2)); -const mono = (sampleRate: number, offsetMs = 0, channel = 0) => ({ - sampleRate, - channels: 1, - channel, - offsetMs, -}); - -test("positioned sources are downmixed and resampled onto separate channels with silent gaps", async () => { - const audio = new PcmAudio({ - sampleRate: 24000, - channels: 2, - }); - audio.update(new Int16Array([100, 300, 300, 500]), { - sampleRate: 12000, - channels: 2, - channel: 0, - offsetMs: 0, - }); - audio.update(new Int16Array([-1, -2]), mono(24000, 2 / 24, 1)); - const result = audio.digest({ ...all, filename: "speech.pcm" }); - expect(samples(await bytes(result))).toEqual([ - 200, 0, 300, 0, 400, -1, 400, -2, - ]); - expect(result).toMatchObject({ - channels: [0, 1], - startOffsetMs: 0, - metadata: { - mime_type: "audio/pcm", - sample_rate_hz: 24000, - channel_count: 2, - sample_format: "s16le", - duration_ms: 4 / 24, - }, - }); - expect(result!.attachment.reference).toMatchObject({ - filename: "speech.pcm", - content_type: "audio/pcm", - }); - expect(audio.bufferedByteLength).toBe(0); -}); - -test("updates copy input and digests cannot be changed by later writes", async () => { - const audio = new PcmAudio({ sampleRate: 1000, channels: 1 }); - const source = new Int16Array([1, 2, 3, 4]); - audio.update(source, mono(1000)); - source.fill(0); - // Later writes win at overlaps. - audio.update(new Int16Array([9, 8]), mono(1000, 1)); - const first = audio.digest({ ...all, endOffsetMs: 2 }); - expect(samples(await bytes(first))).toEqual([1, 9]); - expect(() => audio.update(new Int16Array([7]), mono(1000, 1))).toThrow( - "capture_clock_discontinuity", - ); - const rest = audio.digest(all); - expect(samples(await bytes(rest))).toEqual([8, 4]); - expect(rest!.startOffsetMs).toBe(first!.endOffsetMs); -}); - -test("cuts preserve resampled audio sample-for-sample", async () => { - const whole = new PcmAudio({ sampleRate: 24000, channels: 2 }); - const split = new PcmAudio({ sampleRate: 24000, channels: 2 }); - const source = Int16Array.from( - { length: 4410 }, - (_, i) => ((i * 157) % 60001) - 30000, - ); - for (const audio of [whole, split]) audio.update(source, mono(44100, 0, 1)); - const pieces = []; - for (const endOffsetMs of [7 / 24, 13.5, 33.125, 99, 100]) - pieces.push(await bytes(split.digest({ ...all, endOffsetMs }))); - expect(Buffer.concat(pieces)).toEqual(await bytes(whole.digest(all))); - expect(split.bufferedByteLength).toBe(0); -}); - -test("bounds storage and output allocation without consuming data on failure", async () => { - const audio = new PcmAudio({ - sampleRate: 1000, - channels: 1, - maxBufferBytes: 4, - }); - audio.update(new Int16Array([1, 2]), mono(1000)); - expect(() => audio.update(new Int16Array([3]), mono(1000, 2))).toThrow( - "capture_byte_limit", - ); - expect(audio.bufferedByteLength).toBe(4); - expect(samples(await bytes(audio.digest(all)))).toEqual([1, 2]); - audio.update(new Int16Array([7]), mono(1000, 1000000)); - expect(() => audio.digest(all)).toThrow("capture_byte_limit"); - expect(audio.digest({ ...all, endOffsetMs: 1000000 })).toBeUndefined(); - expect(samples(await bytes(audio.digest(all)))).toEqual([7]); -}); - -test("rejects invalid formats, incomplete frames, cuts and byte limits", () => { - expect(() => new PcmAudio({ sampleRate: 0, channels: 1 })).toThrow(); - expect(() => new PcmAudio({ sampleRate: 24000, channels: 0 })).toThrow(); - const audio = new PcmAudio({ sampleRate: 24000, channels: 2 }); - const stereo = { sampleRate: 24000, channels: 2, channel: 0, offsetMs: 0 }; - expect(() => audio.update(new Int16Array([1]), stereo)).toThrow( - "incomplete_pcm_frame", - ); - expect(() => - audio.update(new Int16Array([1, 2]), { ...stereo, channel: 2 }), - ).toThrow("unsupported_audio_format"); - expect(() => - audio.update(new Int16Array([1, 2]), { ...stereo, offsetMs: NaN }), - ).toThrow("unsupported_audio_format"); - audio.update(new Int16Array([1, 2]), stereo); - for (const endOffsetMs of [1000, -1]) - expect(() => audio.digest({ ...all, endOffsetMs })).toThrow( - "invalid_digest_offset", - ); - for (const maxBytes of [0, -1, 1.5, NaN, Infinity, MAX + 1]) - expect(() => audio.digest({ maxBytes })).toThrow( - "invalid_digest_byte_limit", - ); - expect(audio.bufferedByteLength).toBe(2); -}); - -test("byte-limited digests wait for confirmed full chunks and flush a final remainder", async () => { - const audio = new PcmAudio({ sampleRate: 1000, channels: 2 }); - const source = Int16Array.from({ length: 6 }, (_, i) => i + 1); - audio.update(source, mono(1000)); - expect(audio.digest({ maxBytes: 8, endOffsetMs: 1 })).toBeUndefined(); - const first = audio.digest({ maxBytes: 8, endOffsetMs: 3 }); - expect(first).toMatchObject({ startOffsetMs: 0, endOffsetMs: 2 }); - expect(audio.digest({ maxBytes: 8, endOffsetMs: 3 })).toBeUndefined(); - const second = audio.digest({ maxBytes: 8, endOffsetMs: 3, flush: true }); - expect(second).toMatchObject({ startOffsetMs: 2, endOffsetMs: 3 }); - const third = audio.digest({ maxBytes: 8 }); - expect(audio.digest({ maxBytes: 8 })).toBeUndefined(); - const last = audio.digest({ maxBytes: 8, flush: true }); - const data = Buffer.concat( - await Promise.all([first, second, third, last].map(bytes)), - ); - expect(samples(data)).toEqual([1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6, 0]); - expect(audio.bufferedByteLength).toBe(0); -}); - -test.each([ - { channels: 1, maxBytes: 1, sizes: [2, 2, 2, 2, 2] }, - { channels: 2, maxBytes: 9, sizes: [8, 8, 4] }, - { channels: 3, maxBytes: 13, sizes: [12, 12, 6] }, -])( - "byte limits align to complete frames for $channels channels", - async ({ channels, maxBytes, sizes }) => { - const audio = new PcmAudio({ sampleRate: 1000, channels }); - audio.update(new Int16Array([1, 2, 3, 4, 5]), mono(1000)); - const files: Buffer[] = []; - while (audio.bufferedByteLength) - files.push(await bytes(audio.digest({ maxBytes, flush: true }))); - expect(files.map((f) => f.length)).toEqual(sizes); - }, -); - -test("byte-limited digests skip empty intervals and retain silence inside a file", async () => { - const audio = new PcmAudio({ sampleRate: 1000, channels: 1 }); - audio.update(new Int16Array([7]), mono(1000, 5)); - audio.update(new Int16Array([9]), mono(1000, 8)); - const first = audio.digest({ maxBytes: 4 }); - expect(first).toMatchObject({ startOffsetMs: 4, endOffsetMs: 6 }); - expect(samples(await bytes(first))).toEqual([0, 7]); - expect(audio.digest({ maxBytes: 4 })).toBeUndefined(); - const last = audio.digest({ maxBytes: 4, flush: true }); - expect(last).toMatchObject({ startOffsetMs: 8, endOffsetMs: 9 }); - expect(samples(await bytes(last))).toEqual([9]); -}); diff --git a/js/src/pcm-buffer.test.ts b/js/src/pcm-buffer.test.ts new file mode 100644 index 000000000..d2eb4797c --- /dev/null +++ b/js/src/pcm-buffer.test.ts @@ -0,0 +1,162 @@ +import { expect, test } from "vitest"; +import { PcmBuffer } from "./pcm-buffer"; + +const MAX = Number.MAX_SAFE_INTEGER; +// Without a chunk-sized limit, a partial flush consumes everything through the cut. +const all = { maxBytes: MAX, partial: true }; +function bytes(result: ReturnType) { + return Buffer.from(result!.data); +} +const samples = (data: Buffer) => + Array.from({ length: data.length / 2 }, (_, i) => data.readInt16LE(i * 2)); +const mono = (sampleRate: number, offsetMs = 0, channel = 0) => ({ + sampleRate, + channels: 1, + channel, + offsetMs, +}); + +test("positioned sources are downmixed and resampled onto separate channels with silent gaps", () => { + const audio = new PcmBuffer({ + sampleRate: 24000, + channels: 2, + }); + audio.write(new Int16Array([100, 300, 300, 500]), { + sampleRate: 12000, + channels: 2, + channel: 0, + offsetMs: 0, + }); + audio.write(new Int16Array([-1, -2]), mono(24000, 2 / 24, 1)); + const result = audio.flush(all); + expect(samples(bytes(result))).toEqual([200, 0, 300, 0, 400, -1, 400, -2]); + expect(result).toMatchObject({ + channels: [0, 1], + startOffsetMs: 0, + endOffsetMs: 4 / 24, + }); + expect(audio.bufferedByteLength).toBe(0); +}); + +test("writes copy input and flushes cannot be changed by later writes", () => { + const audio = new PcmBuffer({ sampleRate: 1000, channels: 1 }); + const source = new Int16Array([1, 2, 3, 4]); + audio.write(source, mono(1000)); + source.fill(0); + // Later writes win at overlaps. + audio.write(new Int16Array([9, 8]), mono(1000, 1)); + const first = audio.flush({ ...all, endOffsetMs: 2 }); + expect(samples(bytes(first))).toEqual([1, 9]); + expect(() => audio.write(new Int16Array([7]), mono(1000, 1))).toThrow( + "capture_clock_discontinuity", + ); + const rest = audio.flush(all); + expect(samples(bytes(rest))).toEqual([8, 4]); + expect(rest!.startOffsetMs).toBe(first!.endOffsetMs); +}); + +test("cuts preserve resampled audio sample-for-sample", () => { + const whole = new PcmBuffer({ sampleRate: 24000, channels: 2 }); + const split = new PcmBuffer({ sampleRate: 24000, channels: 2 }); + const source = Int16Array.from( + { length: 4410 }, + (_, i) => ((i * 157) % 60001) - 30000, + ); + for (const audio of [whole, split]) audio.write(source, mono(44100, 0, 1)); + const pieces = []; + for (const endOffsetMs of [7 / 24, 13.5, 33.125, 99, 100]) + pieces.push(bytes(split.flush({ ...all, endOffsetMs }))); + expect(Buffer.concat(pieces)).toEqual(bytes(whole.flush(all))); + expect(split.bufferedByteLength).toBe(0); +}); + +test("bounds storage and output allocation without consuming data on failure", () => { + const audio = new PcmBuffer({ + sampleRate: 1000, + channels: 1, + maxBufferBytes: 4, + }); + audio.write(new Int16Array([1, 2]), mono(1000)); + expect(() => audio.write(new Int16Array([3]), mono(1000, 2))).toThrow( + "capture_byte_limit", + ); + expect(audio.bufferedByteLength).toBe(4); + expect(samples(bytes(audio.flush(all)))).toEqual([1, 2]); + audio.write(new Int16Array([7]), mono(1000, 1000000)); + expect(() => audio.flush(all)).toThrow("capture_byte_limit"); + expect(audio.flush({ ...all, endOffsetMs: 1000000 })).toBeUndefined(); + expect(samples(bytes(audio.flush(all)))).toEqual([7]); +}); + +test("rejects invalid formats, incomplete frames, cuts and byte limits", () => { + expect(() => new PcmBuffer({ sampleRate: 0, channels: 1 })).toThrow(); + expect(() => new PcmBuffer({ sampleRate: 24000, channels: 0 })).toThrow(); + const audio = new PcmBuffer({ sampleRate: 24000, channels: 2 }); + const stereo = { sampleRate: 24000, channels: 2, channel: 0, offsetMs: 0 }; + expect(() => audio.write(new Int16Array([1]), stereo)).toThrow( + "incomplete_pcm_frame", + ); + expect(() => + audio.write(new Int16Array([1, 2]), { ...stereo, channel: 2 }), + ).toThrow("unsupported_audio_format"); + expect(() => + audio.write(new Int16Array([1, 2]), { ...stereo, offsetMs: NaN }), + ).toThrow("unsupported_audio_format"); + audio.write(new Int16Array([1, 2]), stereo); + for (const endOffsetMs of [1000, -1]) + expect(() => audio.flush({ ...all, endOffsetMs })).toThrow( + "invalid_digest_offset", + ); + for (const maxBytes of [0, -1, 1.5, NaN, Infinity, MAX + 1]) + expect(() => audio.flush({ maxBytes })).toThrow( + "invalid_digest_byte_limit", + ); + expect(audio.bufferedByteLength).toBe(2); +}); + +test("byte-limited flushes wait for confirmed full chunks and flush a final remainder", () => { + const audio = new PcmBuffer({ sampleRate: 1000, channels: 2 }); + const source = Int16Array.from({ length: 6 }, (_, i) => i + 1); + audio.write(source, mono(1000)); + expect(audio.flush({ maxBytes: 8, endOffsetMs: 1 })).toBeUndefined(); + const first = audio.flush({ maxBytes: 8, endOffsetMs: 3 }); + expect(first).toMatchObject({ startOffsetMs: 0, endOffsetMs: 2 }); + expect(audio.flush({ maxBytes: 8, endOffsetMs: 3 })).toBeUndefined(); + const second = audio.flush({ maxBytes: 8, endOffsetMs: 3, partial: true }); + expect(second).toMatchObject({ startOffsetMs: 2, endOffsetMs: 3 }); + const third = audio.flush({ maxBytes: 8 }); + expect(audio.flush({ maxBytes: 8 })).toBeUndefined(); + const last = audio.flush({ maxBytes: 8, partial: true }); + const data = Buffer.concat([first, second, third, last].map(bytes)); + expect(samples(data)).toEqual([1, 0, 2, 0, 3, 0, 4, 0, 5, 0, 6, 0]); + expect(audio.bufferedByteLength).toBe(0); +}); + +test.each([ + { channels: 1, maxBytes: 1, sizes: [2, 2, 2, 2, 2] }, + { channels: 2, maxBytes: 9, sizes: [8, 8, 4] }, + { channels: 3, maxBytes: 13, sizes: [12, 12, 6] }, +])( + "byte limits align to complete frames for $channels channels", + ({ channels, maxBytes, sizes }) => { + const audio = new PcmBuffer({ sampleRate: 1000, channels }); + audio.write(new Int16Array([1, 2, 3, 4, 5]), mono(1000)); + const files: Buffer[] = []; + while (audio.bufferedByteLength) + files.push(bytes(audio.flush({ maxBytes, partial: true }))); + expect(files.map((f) => f.length)).toEqual(sizes); + }, +); + +test("byte-limited flushes skip empty intervals and retain silence inside a file", () => { + const audio = new PcmBuffer({ sampleRate: 1000, channels: 1 }); + audio.write(new Int16Array([7]), mono(1000, 5)); + audio.write(new Int16Array([9]), mono(1000, 8)); + const first = audio.flush({ maxBytes: 4 }); + expect(first).toMatchObject({ startOffsetMs: 4, endOffsetMs: 6 }); + expect(samples(bytes(first))).toEqual([0, 7]); + expect(audio.flush({ maxBytes: 4 })).toBeUndefined(); + const last = audio.flush({ maxBytes: 4, partial: true }); + expect(last).toMatchObject({ startOffsetMs: 8, endOffsetMs: 9 }); + expect(samples(bytes(last))).toEqual([9]); +}); diff --git a/js/src/pcm-audio.ts b/js/src/pcm-buffer.ts similarity index 85% rename from js/src/pcm-audio.ts rename to js/src/pcm-buffer.ts index 1728d23cb..f9e959f7d 100644 --- a/js/src/pcm-audio.ts +++ b/js/src/pcm-buffer.ts @@ -1,15 +1,13 @@ -import { Attachment } from "./logger"; - const MAX_CHUNKS = 100_000; type Chunk = { samples: Int16Array; start: number; channel: number }; /** * Incremental signed 16-bit little-endian PCM on a shared sample clock. - * update() copies, resamples, and downmixes a positioned source onto one - * output channel; digest() consumes a prefix into an ordinary attachment. + * write() copies, resamples, and downmixes a positioned source onto one + * output channel; flush() consumes a prefix as interleaved bytes. */ -export class PcmAudio { +export class PcmBuffer { private chunks: Chunk[] = []; private start = 0; private end = 0; @@ -45,8 +43,8 @@ export class PcmAudio { return samplesToMs(this.end, this.sampleRate); } - /** Later writes win where sources overlap; digested audio cannot change. */ - update( + /** Later writes win where sources overlap; flushed audio cannot change. */ + write( data: Int16Array, { sampleRate, @@ -107,15 +105,22 @@ export class PcmAudio { /** * Consume at most maxBytes through an optional confirmed offset; future - * updates may continue. Waits for a full chunk unless flush is set. Cuts align + * writes may continue. Waits for a full chunk unless partial is set. Cuts align * to complete sample frames (at least one), skipping wholly empty intervals. */ - digest(options: { + flush(options: { maxBytes: number; endOffsetMs?: number; - flush?: boolean; - filename?: string; - }) { + partial?: boolean; + }): + | { + data: Uint8Array; + startOffsetMs: number; + endOffsetMs: number; + /** Output channels that received samples. */ + channels: number[]; + } + | undefined { if (!Number.isSafeInteger(options.maxBytes) || options.maxBytes <= 0) throw new Error("invalid_digest_byte_limit"); let end = @@ -132,7 +137,7 @@ export class PcmAudio { // Skip empty chunks without allocating silence or emitting attachments. this.start += Math.floor((first - this.start) / frames) * frames; const cut = this.start + frames; - if (end < cut && !options.flush) return; + if (end < cut && !options.partial) return; end = Math.min(end, cut); const start = this.start; if (end === start) return; @@ -161,25 +166,14 @@ export class PcmAudio { if (chunk.start >= end) return [chunk]; const consumed = Math.min(chunk.samples.length, end - chunk.start); this.bytes -= consumed * 2; - // Copy the tail so it doesn't retain the already digested prefix's allocation. + // Copy the tail so it doesn't retain the already flushed prefix's allocation. return consumed === chunk.samples.length ? [] : [{ ...chunk, start: end, samples: chunk.samples.slice(consumed) }]; }); this.start = end; return { - attachment: new Attachment({ - data: bytes, - filename: options.filename ?? "audio.pcm", - contentType: "audio/pcm", - }), - metadata: { - mime_type: "audio/pcm", - sample_rate_hz: this.sampleRate, - channel_count: this.channels, - sample_format: "s16le", - duration_ms: samplesToMs(end - start, this.sampleRate), - }, + data: bytes, startOffsetMs: samplesToMs(start, this.sampleRate), endOffsetMs: samplesToMs(end, this.sampleRate), channels,