From e2401c807edd4339b9b025cbc4abf9ff78570a73 Mon Sep 17 00:00:00 2001 From: David Elner Date: Thu, 8 Oct 2026 16:54:50 -0400 Subject: [PATCH] Added: Audio recording integration --- .changeset/shared-audio-recording.md | 7 + integrations/audio/README.md | 108 ++++++ integrations/audio/package.json | 51 +++ integrations/audio/smoke/recording.mjs | 51 +++ integrations/audio/src/audio-worker.ts | 46 +++ integrations/audio/src/buffers.ts | 60 +++ integrations/audio/src/codec.ts | 16 + integrations/audio/src/encoder.test.ts | 150 ++++++++ integrations/audio/src/encoder.ts | 121 ++++++ integrations/audio/src/exporter.test.ts | 277 ++++++++++++++ integrations/audio/src/exporter.ts | 61 +++ .../audio/src/fixtures/counting-codec.mjs | 8 + integrations/audio/src/index.ts | 5 + integrations/audio/src/pcm.test.ts | 107 ++++++ integrations/audio/src/pcm.ts | 141 +++++++ integrations/audio/src/publisher.ts | 171 +++++++++ integrations/audio/src/recorder.test.ts | 309 +++++++++++++++ integrations/audio/src/recorder.ts | 288 ++++++++++++++ integrations/audio/src/recording.test.ts | 354 ++++++++++++++++++ integrations/audio/src/recording.ts | 112 ++++++ integrations/audio/src/segment.test.ts | 55 +++ integrations/audio/src/segment.ts | 118 ++++++ integrations/audio/src/selections.test.ts | 68 ++++ integrations/audio/src/selections.ts | 64 ++++ integrations/audio/src/timeline.ts | 23 ++ integrations/audio/src/worker.test.ts | 53 +++ integrations/audio/src/worker.ts | 125 +++++++ integrations/audio/tsconfig.json | 15 + integrations/audio/tsup.config.ts | 13 + pnpm-lock.yaml | 75 ++++ 30 files changed, 3052 insertions(+) create mode 100644 .changeset/shared-audio-recording.md create mode 100644 integrations/audio/README.md create mode 100644 integrations/audio/package.json create mode 100644 integrations/audio/smoke/recording.mjs create mode 100644 integrations/audio/src/audio-worker.ts create mode 100644 integrations/audio/src/buffers.ts create mode 100644 integrations/audio/src/codec.ts create mode 100644 integrations/audio/src/encoder.test.ts create mode 100644 integrations/audio/src/encoder.ts create mode 100644 integrations/audio/src/exporter.test.ts create mode 100644 integrations/audio/src/exporter.ts create mode 100644 integrations/audio/src/fixtures/counting-codec.mjs create mode 100644 integrations/audio/src/index.ts create mode 100644 integrations/audio/src/pcm.test.ts create mode 100644 integrations/audio/src/pcm.ts create mode 100644 integrations/audio/src/publisher.ts create mode 100644 integrations/audio/src/recorder.test.ts create mode 100644 integrations/audio/src/recorder.ts create mode 100644 integrations/audio/src/recording.test.ts create mode 100644 integrations/audio/src/recording.ts create mode 100644 integrations/audio/src/segment.test.ts create mode 100644 integrations/audio/src/segment.ts create mode 100644 integrations/audio/src/selections.test.ts create mode 100644 integrations/audio/src/selections.ts create mode 100644 integrations/audio/src/timeline.ts create mode 100644 integrations/audio/src/worker.test.ts create mode 100644 integrations/audio/src/worker.ts create mode 100644 integrations/audio/tsconfig.json create mode 100644 integrations/audio/tsup.config.ts diff --git a/.changeset/shared-audio-recording.md b/.changeset/shared-audio-recording.md new file mode 100644 index 000000000..bafa618b1 --- /dev/null +++ b/.changeset/shared-audio-recording.md @@ -0,0 +1,7 @@ +--- +"@braintrust/audio": minor +--- + +feat(audio): add shared progressive audio recordings + +Add an optional, framework-independent audio recorder with progressive segmentation, WAV and Ogg/Opus encoding in a worker, and ordinary attachment uploads. Bound source memory and worker admission; publish ready recordings and playback selections only after upload succeeds. Preserve uploaded audio when later capture or trace publication fails. diff --git a/integrations/audio/README.md b/integrations/audio/README.md new file mode 100644 index 000000000..af1a4a263 --- /dev/null +++ b/integrations/audio/README.md @@ -0,0 +1,108 @@ +# Braintrust audio recordings + +Record timestamped audio as progressive attachments, with Ogg/Opus compression by default or optional WAV output. Integrations supply audio frames, timing, and spans. + +## Installation + +```sh +npm install @braintrust/audio +``` + +Requires Node 20+. Installed separately from `braintrust`; enable recording through your integration. + +## Recording API + +Create a recording, supply audio frames, and finish when the session ends: + +```ts +import { Attachment } from "braintrust"; +import { createRecording } from "@braintrust/audio"; + +const recording = createRecording({ + span: sessionSpan, + flush: () => logger.flush(), + createAttachment: (options) => new Attachment(options), + audioFormat: "ogg", // Default; use "wav" for uncompressed audio. + options: { segmentDurationSeconds: 60 }, + snapshot: () => ({ + origin: sessionStartUnixMs, + basis: "local_capture", + source: (channel_index) => ({ channel_index }), + }), + turnSelections: () => [], // Add turn playback selections as shown below. +}); + +recording.record({ + pcm, // Int16Array; the recorder copies it. + rate: 24000, + channels: 1, + channel: 0, // Caller; channel 1 is the agent. + at: frameOffsetMs, // Milliseconds since sessionStartUnixMs. +}); +// No more audio will arrive before this offset, on either channel. +recording.advance(confirmedThroughMs); + +// Finish uploads and publish final recording states and selections. +await recording.finish(); +``` + +## Configuration + +Required inputs: + +| Option | Purpose | +| ------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `span` | Session span that receives recording metadata and audio attachments. | +| `createAttachment` | Creates an uploadable attachment; typically `(options) => new Attachment(options)`. | +| `flush` | Flushes trace updates; typically `() => logger.flush()`. | +| `snapshot` | Returns the current timing and source information: `{ origin, basis, source, metadata? }`. `origin` is the session start in Unix milliseconds; `source(channel)` describes each channel. | +| `turnSelections` | Returns `{ span, intervals }` entries linking turns to their audio. Use `() => []` when no turn playback links are needed. | + +Optional settings: + +| Option | Default | Meaning | +| -------------------------------- | ----------- | -------------------------------------------------------------------------- | +| `audioFormat` | `"ogg"` | `"ogg"` for compression; `"wav"` for uncompressed audio. | +| `options.segmentDurationSeconds` | `60` | Target duration per file; buffer pressure can produce shorter files. | +| `options.maxDurationSeconds` | `1800` | Stop capture at this session offset in seconds. | +| `options.maxBufferBytes` | `33554432` | Maximum retained PCM bytes per recording. | +| `options.flushFraction` | `0.5` | Fraction of the buffer limit that triggers an earlier segment. | +| `encoder` | `undefined` | Custom encoder `{ module: "..." }`; overrides `audioFormat` when supplied. | + +## Turn playback + +When reviewing a voice trace, you may want to follow a message, e.g. such as “Where is my order?”, and hear how it was spoken, without searching through the call recording. These selections describe which audio to play for that message. + +Supply `turnSelections` to `createRecording()`: a callback that returns each turn’s span and when its audio occurred, in milliseconds from the start of the call. The recorder maps those timestamps onto the uploaded audio files, even when a turn crosses a file boundary, and writes `audio.selections` onto the turn’s span. These references can be used to play that turn’s audio. + +For example, if the caller said “Where is my order?” from 1.2 to 2.6 seconds into the call: + +```ts +// Inside the createRecording({ ... }) options: +turnSelections: () => [ + { + span: userTurnSpan, // The span containing “Where is my order?” + intervals: [ + { + recording_span_id: sessionSpan.spanId, // Where the call audio is attached. + recording_id: "call", + channel_index: 0, // Caller audio. + start_offset_ms: 1200, + end_offset_ms: 2600, + }, + ], + }, +]; +``` + +When a turn or its timing changes, update what `turnSelections()` returns and call `recording.publishSelections()`. `finish()` also publishes the final selections. These selections associate recorded audio with turns; they do not control what gets recorded. + +## Development + +```sh +pnpm run build +pnpm run check:typings +pnpm test +pnpm run test:smoke +pnpm pack +``` diff --git a/integrations/audio/package.json b/integrations/audio/package.json new file mode 100644 index 000000000..f4cdd7f81 --- /dev/null +++ b/integrations/audio/package.json @@ -0,0 +1,51 @@ +{ + "name": "@braintrust/audio", + "version": "0.0.1", + "description": "Shared audio recording, encoding, and export for Braintrust integrations", + "main": "./dist/index.js", + "module": "./dist/index.mjs", + "types": "./dist/index.d.ts", + "files": [ + "dist", + "README.md" + ], + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.mjs", + "require": "./dist/index.js" + }, + "./package.json": "./package.json" + }, + "scripts": { + "build": "tsup", + "check:typings": "tsc --noEmit", + "test": "vitest run", + "test:smoke": "node smoke/recording.mjs" + }, + "dependencies": { + "@audio/encode-opus": "1.3.1" + }, + "devDependencies": { + "@types/node": "^20.10.5", + "tsup": "^8.5.1", + "typescript": "^5.3.3", + "vitest": "4.1.11", + "ogg-opus-decoder": "^1.7.3" + }, + "engines": { + "node": ">=20" + }, + "author": "Braintrust Data Inc.", + "license": "MIT", + "repository": { + "type": "git", + "url": "git+https://github.com/braintrustdata/braintrust-sdk-javascript.git", + "directory": "integrations/audio" + }, + "publishConfig": { + "access": "public", + "registry": "https://registry.npmjs.org/", + "provenance": true + } +} diff --git a/integrations/audio/smoke/recording.mjs b/integrations/audio/smoke/recording.mjs new file mode 100644 index 000000000..0a40660c4 --- /dev/null +++ b/integrations/audio/smoke/recording.mjs @@ -0,0 +1,51 @@ +import assert from "node:assert/strict"; +import { createRequire } from "node:module"; + +// Also run this file from an isolated app installed from `pnpm pack` output. +const require = createRequire(import.meta.url); +for (const api of [ + await import("@braintrust/audio"), + require("@braintrust/audio"), +]) { + for (const audioFormat of ["wav", "ogg"]) { + const rows = []; + const files = []; + const recording = api.createRecording({ + audioFormat, + span: { spanId: "session", log: (row) => rows.push(row) }, + flush: async () => {}, + createAttachment: ({ data, filename, contentType }) => { + files.push(data); + return { + reference: { key: filename, filename, content_type: contentType }, + upload: async () => ({ upload_status: "done" }), + }; + }, + snapshot: () => ({ + origin: 1000, + basis: "test", + + source: (channel_index) => ({ channel_index }), + }), + turnSelections: () => [], + }); + recording.record({ + pcm: new Int16Array(480).fill(1000), + at: 0, + rate: 24000, + channels: 1, + channel: 0, + }); + await recording.finish(); + assert.equal(files.length, 1); + assert.equal( + Buffer.from(files[0].subarray(0, 4)).toString(), + audioFormat === "wav" ? "RIFF" : "OggS", + ); + assert.equal( + rows.flatMap((r) => r.metadata?.["audio.recordings"] ?? []).at(-1).state, + "ready", + ); + } +} +console.log("WAV and Ogg recording passed through ESM and CommonJS exports."); diff --git a/integrations/audio/src/audio-worker.ts b/integrations/audio/src/audio-worker.ts new file mode 100644 index 000000000..a7e4acb4c --- /dev/null +++ b/integrations/audio/src/audio-worker.ts @@ -0,0 +1,46 @@ +import { parentPort } from "node:worker_threads"; +import { encodeCall, pcmChunks } from "./pcm"; +import type { Packet } from "./recorder"; +parentPort?.on( + "message", + async ({ + packets, + durationMs, + encoderModule, + }: { + packets: Packet[]; + durationMs: number; + encoderModule?: string; + }) => { + try { + const codec = encoderModule ? await import(encoderModule) : undefined; + if (codec?.encodeChunks) { + // The optional codec preserves resampling history between PCM blocks. + const encoded = await codec.encodeChunks( + pcmChunks(packets, durationMs), + 24000, + 2, + ); + parentPort!.postMessage( + { ...encoded, durationMs: Math.round(durationMs * 24) / 24 }, + [encoded.bytes.buffer], + ); + } else { + // Preserve compatibility with codecs exposing the original encode API. + const result = encodeCall(packets, durationMs); + const encoded = codec + ? await codec.encode( + new Int16Array(result.bytes.buffer, 44), + 24000, + 2, + ) + : { bytes: result.bytes, mimeType: "audio/wav", sampleRate: 24000 }; + parentPort!.postMessage({ ...encoded, durationMs: result.durationMs }, [ + encoded.bytes.buffer, + ]); + } + } catch { + parentPort!.postMessage({ reason: "encoding_failed" }); + } + }, +); diff --git a/integrations/audio/src/buffers.ts b/integrations/audio/src/buffers.ts new file mode 100644 index 000000000..d01e139a3 --- /dev/null +++ b/integrations/audio/src/buffers.ts @@ -0,0 +1,60 @@ +/** Owns PCM reservations across staging, recorder admission and worker transfer. */ +export class PcmBuffers { + private leases = new Map< + ArrayBufferLike, + { bytes: number; admitted: boolean } + >(); + private retained = 0; + constructor(private maximum: number) {} + get bytes() { + return this.retained; + } + copy(pcm: Int16Array): Int16Array { + const bytes = pcm.byteLength; + if (this.retained + bytes > this.maximum) { + throw new Error("capture_byte_limit"); + } + const owned = new Int16Array(pcm); + this.leases.set(owned.buffer, { bytes, admitted: false }); + this.retained += bytes; + return owned; + } + + admit(pcm: Int16Array): boolean { + const lease = this.leases.get(pcm.buffer); + if (!lease || lease.admitted) { + return false; + } + lease.admitted = true; + return true; + } + discard(pcm: Int16Array) { + if (!this.leases.get(pcm.buffer)?.admitted) { + this.release(pcm); + } + } + release(pcm: Int16Array) { + const lease = this.leases.get(pcm.buffer); + if (!lease) { + return; + } + this.leases.delete(pcm.buffer); + this.retained -= lease.bytes; + } + split(pcm: Int16Array, count: number): [Int16Array, Int16Array] { + const lease = this.leases.get(pcm.buffer); + if (!lease || pcm.byteLength !== lease.bytes) { + throw new Error("Invalid PCM ownership"); + } + const head = pcm.slice(0, count), + tail = pcm.slice(count); + this.leases.delete(pcm.buffer); + this.leases.set(head.buffer, { bytes: head.byteLength, admitted: true }); + this.leases.set(tail.buffer, { bytes: tail.byteLength, admitted: true }); + return [head, tail]; + } + close() { + this.leases.clear(); + this.retained = 0; + } +} diff --git a/integrations/audio/src/codec.ts b/integrations/audio/src/codec.ts new file mode 100644 index 000000000..2eff29bd7 --- /dev/null +++ b/integrations/audio/src/codec.ts @@ -0,0 +1,16 @@ +import { createRequire } from "node:module"; +import { dirname, join } from "node:path"; +import { pathToFileURL } from "node:url"; +/** Codec is loaded inside Braintrust's encoding worker, never on the voice thread. */ +export function oggOpus() { + return { + module: pathToFileURL( + join( + dirname( + createRequire(__filename).resolve("@braintrust/audio/package.json"), + ), + "dist/encoder.mjs", + ), + ).href, + }; +} diff --git a/integrations/audio/src/encoder.test.ts b/integrations/audio/src/encoder.test.ts new file mode 100644 index 000000000..3e6329340 --- /dev/null +++ b/integrations/audio/src/encoder.test.ts @@ -0,0 +1,150 @@ +import { expect, test } from "vitest"; +import { OggOpusDecoder } from "ogg-opus-decoder"; +import { encode, encodeChunks } from "./encoder"; +import opus from "@audio/encode-opus"; +test("independently decodable Ogg segments preserve channels, length and reduce speech-like PCM size", async () => { + for (const length of [5280, 24000, 24001, 24000 * 3 + 123]) { + const pcm = new Int16Array(length * 2); + for (let i = 0; i < length; i++) { + pcm[i * 2] = Math.round( + Math.sin((i / 24000) * 2 * Math.PI * 220) * 10000, + ); + pcm[i * 2 + 1] = Math.round( + Math.sin((i / 24000) * 2 * Math.PI * 440) * 5000, + ); + } + const encoded = await encode(pcm, 24000, 2); + expect(Buffer.from(encoded.bytes.subarray(0, 4)).toString()).toBe("OggS"); + const decoder = new OggOpusDecoder(); + await decoder.ready; + try { + const decoded = await decoder.decodeFile(encoded.bytes); + expect(decoded.errors).toHaveLength(0); + expect(decoded.channelData).toHaveLength(2); + const data = Buffer.from(encoded.bytes); + const pages: number[] = []; + for (let at = 0; at < data.length; ) { + expect(data.toString("ascii", at, at + 4)).toBe("OggS"); + pages.push(at); + const count = data[at + 26]; + at += + 27 + + count + + data + .subarray(at + 27, at + 27 + count) + .reduce((sum, n) => sum + n, 0); + } + const last = pages.at(-1)!; + const preSkip = data.readUInt16LE(27 + data[26] + 10); + expect(data[last + 5] & 4).toBe(4); + expect(Number(data.readBigUInt64LE(last + 6)) - preSkip).toBe(length * 2); + // ogg-opus-decoder 1.7.3 incorrectly retains pre-skip in its end trim + // when the only audio page is EOS. libsndfile independently decodes + // the 220ms case to its exact duration. Restrict the known + // decoder discrepancy to that layout, never allow missing samples. + if (pages.length === 3) { + expect([length * 2, length * 2 + preSkip]).toContain( + decoded.samplesDecoded, + ); + } else { + expect(decoded.samplesDecoded).toBe(length * 2); + } + expect( + Math.max(...decoded.channelData[0].subarray(1000, 2000)), + ).toBeGreaterThan(0.2); + expect(encoded.bytes.length).toBeLessThan(pcm.byteLength / 3); + } finally { + decoder.free(); + } + } +}); + +test("streaming native-rate blocks decode identically across processing boundaries and the final tail", async () => { + const { encodeChunks } = await import("./encoder"); + const pcm = new Int16Array((48000 * 6 + 17) * 2); + for (let i = 0; i < pcm.length / 2; i++) { + pcm[i * 2] = Math.round(11000 * Math.sin((i * Math.PI * 2 * 317) / 48000)); + pcm[i * 2 + 1] = + i < 48000 || i > 48000 * 4 + ? 0 + : Math.round(6000 * Math.sin((i * Math.PI * 2 * 571) / 48000)); + } + function* chunks() { + for (let i = 0; i < pcm.length; i += 1234) { + yield pcm.subarray(i, i + 1234); + } + } + const complete = await encode(pcm, 48000, 2), + streamed = await encodeChunks(chunks(), 48000, 2); + const decoder = new OggOpusDecoder(); + await decoder.ready; + try { + const reference = await decoder.decodeFile(complete.bytes); + await decoder.reset(); + const result = await decoder.decodeFile(streamed.bytes); + expect(result.errors).toEqual([]); + expect(result.samplesDecoded).toBe(pcm.length / 2); + for (let channel = 0; channel < 2; channel++) { + expect( + Buffer.from(result.channelData[channel].buffer).equals( + Buffer.from(reference.channelData[channel].buffer), + ), + ).toBe(true); + } + } finally { + decoder.free(); + } +}); + +test("streamed resampling preserves whole-file encoding across boundaries and short tails", async () => { + for (const length of [1, 2, 3, 24001, 24002, 24003, 72123]) { + const pcm = new Int16Array(length * 2); + const planar = [new Float32Array(length), new Float32Array(length)]; + for (let i = 0; i < length; i++) { + for (let c = 0; c < 2; c++) { + pcm[i * 2 + c] = Math.round(9000 * Math.sin(i * (c ? 1.7 : 0.13))); + planar[c][i] = pcm[i * 2 + c] / 32768; + } + } + const original = await opus({ + sampleRate: 24000, + channels: 2, + bitrate: 64, + complexity: 5, + application: "audio", + }); + let referenceBytes: Uint8Array; + try { + referenceBytes = Buffer.concat([ + original.encode(planar), + original.flush(), + ]); + } finally { + original.free(); + } + function* chunks() { + for (let i = 0; i < pcm.length; i += 4800) { + yield pcm.subarray(i, i + 4800); + } + } + const streamed = await encodeChunks(chunks(), 24000, 2); + const decoder = new OggOpusDecoder(); + await decoder.ready; + try { + const reference = await decoder.decodeFile(referenceBytes); + await decoder.reset(); + const result = await decoder.decodeFile(streamed.bytes); + expect(result.errors).toEqual([]); + expect(result.samplesDecoded).toBe(reference.samplesDecoded); + for (let c = 0; c < 2; c++) { + expect( + Buffer.from(result.channelData[c].buffer).equals( + Buffer.from(reference.channelData[c].buffer), + ), + ).toBe(true); + } + } finally { + decoder.free(); + } + } +}); diff --git a/integrations/audio/src/encoder.ts b/integrations/audio/src/encoder.ts new file mode 100644 index 000000000..23dda1e19 --- /dev/null +++ b/integrations/audio/src/encoder.ts @@ -0,0 +1,121 @@ +import opus from "@audio/encode-opus"; +import { toOpusRate } from "@audio/encode-opus/core"; + +export async function encode( + pcm: Int16Array, + sampleRate: number, + channels: number, +) { + function* chunks() { + const size = (sampleRate / 10) * channels; + for (let at = 0; at < pcm.length; at += size) { + yield pcm.subarray(at, at + size); + } + } + return encodeChunks(chunks(), sampleRate, channels); +} + +/** Preserve the codec's Lanczos filter across blocks using three samples of + * history and lookahead. Only the beginning/end of the file clamp the filter. */ +function* planarBlocks( + chunks: Iterable, + sampleRate: number, + channels: number, +) { + if (![24000, 48000].includes(sampleRate)) { + throw new Error("unsupported_codec_rate"); + } + const tail = Array.from({ length: channels }, () => new Float32Array(6)); + let retained = 3, + initialized = false; + let input: Float32Array[] = [], + output: Float32Array[] = []; + function convert(length: number, end: number) { + const interleaved = toOpusRate( + input.map((c) => c.subarray(0, length)), + sampleRate, + ); + const ratio = 48000 / sampleRate, + count = (end - 3) * ratio; + if (!output.length || output[0].length < count) { + output = Array.from({ length: channels }, () => new Float32Array(count)); + } + for (let i = 0; i < count; i++) { + for (let c = 0; c < channels; c++) { + output[c][i] = interleaved[(i + 3 * ratio) * channels + c]; + } + } + return output.map((c) => c.subarray(0, count)); + } + for (const pcm of chunks) { + if (!pcm.length) { + continue; + } + const length = retained + pcm.length / channels; + if (!input.length || input[0].length < length) { + input = Array.from({ length: channels }, () => new Float32Array(length)); + } + if (!initialized) { + for (let c = 0; c < channels; c++) { + tail[c].fill(pcm[c] / 32768, 0, 3); + } + initialized = true; + } + for (let c = 0; c < channels; c++) { + input[c].set(tail[c].subarray(0, retained)); + for (let i = 0; i < pcm.length / channels; i++) { + input[c][retained + i] = pcm[i * channels + c] / 32768; + } + } + const end = Math.max(3, length - 3); + if (end > 3) { + yield convert(length, end); + } + retained = length - end + 3; + for (let c = 0; c < channels; c++) { + tail[c].set(input[c].subarray(end - 3, length)); + } + } + if (initialized) { + for (let c = 0; c < channels; c++) { + input[c].set(tail[c].subarray(0, retained)); + } + yield convert(retained, retained); + } +} + +/** Worker-only streaming codec. Blocks are consumed before advancing input. */ +export async function encodeChunks( + chunks: Iterable, + sampleRate: number, + channels: number, +) { + const encoder = await opus({ + sampleRate: 48000, + channels, + bitrate: 64, + complexity: 5, + application: "audio", + }); + try { + const pages: Uint8Array[] = []; + let length = 0; + for (const planar of planarBlocks(chunks, sampleRate, channels)) { + const page = encoder.encode(planar); + pages.push(page); + length += page.length; + } + const last = encoder.flush(); + pages.push(last); + length += last.length; + const bytes = new Uint8Array(length); + let offset = 0; + for (const page of pages) { + bytes.set(page, offset); + offset += page.length; + } + return { bytes, mimeType: "audio/ogg", sampleRate: 48000 }; + } finally { + encoder.free(); + } +} diff --git a/integrations/audio/src/exporter.test.ts b/integrations/audio/src/exporter.test.ts new file mode 100644 index 000000000..28b3cbb31 --- /dev/null +++ b/integrations/audio/src/exporter.test.ts @@ -0,0 +1,277 @@ +import { afterEach, expect, test, vi } from "vitest"; +import { AudioSegment } from "./segment"; +import { SegmentExporter } from "./exporter"; +import { TracePublisher } from "./publisher"; +class TestAttachment { + reference: { key: string; filename: string; content_type: string }; + constructor(options: { filename: string; contentType: string }) { + this.reference = { + key: "attachment", + filename: options.filename, + content_type: options.contentType, + }; + } + async upload(): Promise<{ + upload_status: "done" | "error"; + error_message?: string; + }> { + return { upload_status: "done" }; + } +} +afterEach(() => vi.restoreAllMocks()); + +function fixture() { + const span = { spanId: "root", log: vi.fn() }; + const turn = { log: vi.fn() }, + speaking = { log: vi.fn() }; + const segments: AudioSegment[] = []; + const flush = vi.fn(async () => {}); + const intervals = [ + { + recording_span_id: "root", + recording_id: "call", + channel_index: 1, + start_offset_ms: 100, + end_offset_ms: 140, + }, + ]; + const publisher = new TracePublisher( + span, + flush, + () => ({ + segments, + origin: 1234, + basis: "test", + source: (channel) => ({ channel_index: channel }), + closed: false, + }), + () => [ + { span: turn, intervals, alias: true }, + { span: speaking, intervals }, + ], + ); + const executions = new Map(); + const exporter = { + publishRecordings: () => publisher.publishRecordings(), + publishSelections: () => publisher.publishSelections(), + publish: async (segment: AudioSegment) => { + await executions.get(segment)!.export(segment, () => {}); + await publisher.publish(segment); + }, + }; + function encode(error?: string, bytes = 16) { + const start = 100 + segments.length * 20; + const segment = new AudioSegment( + `call-${segments.length}`, + start, + start + 20, + [ + { + pcm: new Int16Array(480), + at: 0, + rate: 24000, + channels: 1, + channel: 1, + }, + ], + ); + segments.push(segment); + publisher.publishRecordings(); + executions.set( + segment, + new SegmentExporter( + async () => { + if (error) { + throw new Error(error); + } + return { + bytes: new Uint8Array(bytes), + durationMs: 20, + sampleRate: 24000, + mimeType: "audio/wav", + }; + }, + (options) => new TestAttachment(options), + ), + ); + return segment; + } + const manifests = () => + span.log.mock.calls.flatMap( + ([row]) => row.metadata?.["audio.recordings"] ?? [], + ); + return { span, turn, speaking, exporter, encode, manifests }; +} + +test.each([16, 17 * 1024 * 1024])( + "%i encoded bytes remain pending until ordinary attachment upload succeeds", + async (bytes) => { + let complete!: (status: { upload_status: "done" }) => void; + const upload = vi + .spyOn(TestAttachment.prototype, "upload") + .mockImplementation( + () => + new Promise((resolve) => { + complete = resolve; + }), + ); + const f = fixture(), + segment = f.encode(undefined, bytes); + const publication = f.exporter.publish(segment); + try { + f.exporter.publishRecordings(); + f.exporter.publishSelections(); + expect(f.manifests().every((m) => m.state === "pending")).toBe(true); + expect(f.span.log.mock.calls.some(([row]) => row.input?.audio)).toBe( + false, + ); + for (const owner of [f.turn, f.speaking]) { + expect( + owner.log.mock.calls.flatMap( + ([row]) => row.metadata?.["audio.selections"] ?? [], + ), + ).toEqual([]); + } + await vi.waitFor(() => expect(upload).toHaveBeenCalledTimes(1)); + complete({ upload_status: "done" }); + await publication; + expect(f.manifests().at(-1)).toMatchObject({ + state: "ready", + attachment: { span_id: "root", ref: "/input/audio/call-0" }, + }); + expect(f.manifests()[0].state).toBe("pending"); + for (const owner of [f.turn, f.speaking]) { + expect( + owner.log.mock.calls.at(-1)![0].metadata["audio.selections"], + ).toMatchObject([ + { recording_id: "call-0", start_offset_ms: 0, end_offset_ms: 20 }, + ]); + } + const reference = f.span.log.mock.calls.find( + ([row]) => row.input?.audio, + )![0].input.audio[segment.id]; + expect(reference).toMatchObject({ + key: expect.any(String), + filename: "call-0.wav", + content_type: "audio/wav", + }); + expect(reference).not.toBeInstanceOf(TestAttachment); + const rows = f.span.log.mock.calls.length; + f.exporter.publishRecordings(); + f.exporter.publishSelections(); + expect(f.span.log.mock.calls).toHaveLength(rows); + } finally { + complete?.({ upload_status: "done" }); + await publication.catch(() => {}); + } + }, +); + +test.each(["status", "rejection"])( + "upload %s failure never publishes ready and preserves earlier audio", + async (mode) => { + const upload = vi + .spyOn(TestAttachment.prototype, "upload") + .mockResolvedValueOnce({ upload_status: "done" }); + const f = fixture(); + await f.exporter.publish(f.encode()); + if (mode === "status") { + upload.mockResolvedValue({ + upload_status: "error", + error_message: "upload unavailable", + }); + } else { + upload.mockRejectedValue(new Error("upload unavailable")); + } + await f.exporter.publish(f.encode()); + expect( + f + .manifests() + .filter((m) => m.id === "call-1") + .map((m) => m.state), + ).not.toContain("ready"); + expect(f.manifests().at(-1)).not.toHaveProperty("attachment"); + expect( + f.span.log.mock.calls.flatMap(([row]) => + Object.keys(row.input?.audio ?? {}), + ), + ).toEqual(["call-0"]); + expect(f.manifests().at(-1)).toMatchObject({ + id: "call-1", + state: "omitted", + reason: "recording_export_failed", + }); + expect( + f + .manifests() + .filter((m) => m.id === "call-0") + .at(-1).state, + ).toBe("ready"); + for (const owner of [f.turn, f.speaking]) { + expect( + owner.log.mock.calls.at(-1)![0].metadata["audio.selections"], + ).toMatchObject([{ recording_id: "call-0" }]); + } + }, +); + +test("encoding failure publishes an omission without uploading or selecting audio", async () => { + const upload = vi.spyOn(TestAttachment.prototype, "upload"); + const f = fixture(), + segment = f.encode("encoding_failed"); + await f.exporter.publish(segment); + expect(upload).not.toHaveBeenCalled(); + expect(f.manifests().map((m) => m.state)).toEqual(["pending", "omitted"]); + expect(f.manifests().at(-1)).toMatchObject({ reason: "encoding_failed" }); + expect(f.span.log.mock.calls.some(([row]) => row.input?.audio)).toBe(false); + expect(f.turn.log.mock.calls.at(-1)![0].metadata["audio.selections"]).toEqual( + [], + ); +}); + +test.each([false, true])( + "encoding failure=%s releases PCM only after encoding settles", + async (failed) => { + let finish!: () => void; + const encoding = new Promise((resolve) => { + finish = resolve; + }); + const pcm = new Int16Array(480); + const segment = new AudioSegment("call-0", 0, 20, [ + { pcm, at: 0, rate: 24000, channels: 1, channel: 0 }, + ]); + const release = vi.fn(); + const upload = vi.fn(async () => { + expect(release).toHaveBeenCalledExactlyOnceWith(pcm); + expect(segment.state).toBe("pending"); + return { upload_status: "done" }; + }); + const exporter = new SegmentExporter( + async () => { + await encoding; + if (failed) { + throw new Error("encoding_failed"); + } + return { + bytes: new Uint8Array(16), + durationMs: 20, + sampleRate: 24000, + mimeType: "audio/wav", + }; + }, + () => ({ reference: "uploaded", upload }), + ); + const job = exporter.export(segment, release); + try { + expect(release).not.toHaveBeenCalled(); + expect(upload).not.toHaveBeenCalled(); + expect(segment.state).toBe("pending"); + } finally { + finish(); + await job; + } + expect(release).toHaveBeenCalledExactlyOnceWith(pcm); + expect(upload).toHaveBeenCalledTimes(failed ? 0 : 1); + expect(segment.state).toBe(failed ? "omitted" : "ready"); + }, +); diff --git a/integrations/audio/src/exporter.ts b/integrations/audio/src/exporter.ts new file mode 100644 index 000000000..276d9e097 --- /dev/null +++ b/integrations/audio/src/exporter.ts @@ -0,0 +1,61 @@ +import { AudioSegment, type Packet, type EncodedAudio } from "./segment"; + +export type Encoder = ( + packets: Packet[], + durationMs: number, +) => Promise; +export type CreateAttachment = (options: { + data: Uint8Array; + filename: string; + contentType: string; +}) => { + upload(): Promise<{ upload_status: string; error_message?: string }>; + reference: unknown; +}; + +/** Executes a sealed segment; recording policy and trace publication stay with callers. */ +export class SegmentExporter { + constructor( + private encode: Encoder, + private createAttachment: CreateAttachment, + ) {} + + async export( + segment: AudioSegment, + release: (pcm: Int16Array) => void, + ): Promise { + const packets = segment.dispatch({ type: "encode" }); + let data: EncodedAudio; + try { + data = await this.encode(packets, segment.end - segment.start); + segment.dispatch({ type: "encoded", format: data }); + } catch (error) { + segment.dispatch({ + type: "omitted", + reason: error instanceof Error ? error.message : "encoding_failed", + }); + return; + } finally { + // Release source memory only after the worker has finished with it. + for (const packet of packets) { + release(packet.pcm); + } + packets.length = 0; + } + + try { + const attachment = this.createAttachment({ + data: data.bytes, + filename: `${segment.id}.${data.mimeType === "audio/ogg" ? "ogg" : "wav"}`, + contentType: data.mimeType, + }); + const result = await attachment.upload(); + if (result.upload_status !== "done") { + throw new Error("recording_export_failed"); + } + segment.dispatch({ type: "ready", reference: attachment.reference }); + } catch { + segment.dispatch({ type: "omitted", reason: "recording_export_failed" }); + } + } +} diff --git a/integrations/audio/src/fixtures/counting-codec.mjs b/integrations/audio/src/fixtures/counting-codec.mjs new file mode 100644 index 000000000..61f93cf8b --- /dev/null +++ b/integrations/audio/src/fixtures/counting-codec.mjs @@ -0,0 +1,8 @@ +let calls = 0; +export function encode() { + return { + bytes: Uint8Array.of(++calls), + mimeType: "audio/wav", + sampleRate: 24000, + }; +} diff --git a/integrations/audio/src/index.ts b/integrations/audio/src/index.ts new file mode 100644 index 000000000..7b22e98b4 --- /dev/null +++ b/integrations/audio/src/index.ts @@ -0,0 +1,5 @@ +export { createRecording, type RecordingConfig } from "./recording"; +export { oggOpus } from "./codec"; +export type { RecordingOptions, Packet } from "./recorder"; +export type { Selection } from "./selections"; +export * as timeline from "./timeline"; diff --git a/integrations/audio/src/pcm.test.ts b/integrations/audio/src/pcm.test.ts new file mode 100644 index 000000000..31cec3f0e --- /dev/null +++ b/integrations/audio/src/pcm.test.ts @@ -0,0 +1,107 @@ +import { expect, test } from "vitest"; +import { encodeCall, pcmChunks } from "./pcm"; + +test("bounded rendering preserves resampling phase across chunks and pads the final timeline", () => { + const source = Int16Array.from( + { length: 12000 }, + (_, i) => (i % 97) * 100 - 4000, + ); + for (const rate of [24000, 48000]) { + const frames = Array.from( + pcmChunks( + [{ pcm: source, rate: 44100, channels: 1, channel: 1, at: 7.3 }], + 400, + rate, + ), + (block) => block.slice(), + ); + expect(Math.max(...frames.map((f) => f.length))).toBeLessThanOrEqual( + (rate / 10) * 2, + ); + const actual = Int16Array.from(frames.flatMap((f) => [...f])); + expect(actual.length).toBe(rate * 0.4 * 2); + const start = Math.round((7.3 * rate) / 1000), + length = Math.round((source.length / 44100) * rate); + for (let i = 0; i < actual.length / 2; i++) { + const position = ((i - start) * 44100) / rate, + lo = Math.floor(position), + alpha = position - lo; + const expected = + i < start || i >= start + length + ? 0 + : Math.round( + source[Math.min(lo, source.length - 1)] * (1 - alpha) + + source[Math.min(lo + 1, source.length - 1)] * alpha, + ); + expect(actual[i * 2]).toBe(0); + expect(actual[i * 2 + 1]).toBe(expected); + } + } +}); + +test("rendering preserves packet overwrite order, channel separation, silence and WAV samples", () => { + const packets = [ + { + pcm: new Int16Array(4800).fill(123), + rate: 24000, + channels: 1, + channel: 0, + at: 50, + }, + { + pcm: new Int16Array(4800).fill(456), + rate: 24000, + channels: 1, + channel: 1, + at: 0, + }, + { + pcm: new Int16Array(4800).fill(789), + rate: 24000, + channels: 1, + channel: 0, + at: 0, + }, + ]; + const pcm = Int16Array.from( + Array.from(pcmChunks(packets, 300), (block) => [...block]).flat(), + ); + const wav = encodeCall(packets, 300), + view = new DataView(wav.bytes.buffer); + for (let i = 0; i < pcm.length / 2; i++) { + expect(pcm[i * 2]).toBe(i < 4800 ? 789 : i < 6000 ? 123 : 0); + expect(pcm[i * 2 + 1]).toBe(i < 4800 ? 456 : 0); + expect(view.getInt16(44 + i * 4, true)).toBe(pcm[i * 2]); + expect(view.getInt16(46 + i * 4, true)).toBe(pcm[i * 2 + 1]); + } +}); + +// Exercise both upsampling and downsampling through the rendered WAV output. +test("mixed source rates retain channel values and duration", () => { + const result = encodeCall([ + { + pcm: new Int16Array(160).fill(1000), + rate: 8000, + channels: 1, + channel: 0, + at: 0, + }, + { + pcm: new Int16Array(960).fill(-1000), + rate: 48000, + channels: 1, + channel: 1, + at: 0, + }, + ]); + expect(result.durationMs).toBe(20); + const view = new DataView( + result.bytes.buffer, + result.bytes.byteOffset, + result.bytes.byteLength, + ); + for (let offset = 44; offset < result.bytes.byteLength; offset += 4) { + expect(view.getInt16(offset, true)).toBe(1000); + expect(view.getInt16(offset + 2, true)).toBe(-1000); + } +}); diff --git a/integrations/audio/src/pcm.ts b/integrations/audio/src/pcm.ts new file mode 100644 index 000000000..b01f49cbc --- /dev/null +++ b/integrations/audio/src/pcm.ts @@ -0,0 +1,141 @@ +import { msToSamples, samplesToMs, CALL_SAMPLE_RATE } from "./timeline"; +import type { Packet } from "./recorder"; +/** Bounded 24kHz stereo WAV assembly, entirely outside the voice event loop. */ +export function encodeCall( + packets: Packet[], + durationMs?: number, +): { + bytes: Uint8Array; + durationMs: number; +} { + const rate = CALL_SAMPLE_RATE; + const samples = + durationMs === undefined + ? Math.ceil( + packets.reduce( + (end, p) => + Math.max( + end, + msToSamples(p.at, rate) + + (p.pcm.length / p.channels / p.rate) * rate, + ), + 0, + ), + ) + : Math.round(msToSamples(durationMs, rate)); + if (!Number.isSafeInteger(samples) || samples <= 0 || samples > 1800 * rate) { + throw new Error("invalid_duration"); + } + const bytes = new Uint8Array(44 + samples * 4); + const view = new DataView(bytes.buffer); + const text = (offset: number, value: string) => { + for (let i = 0; i < value.length; i++) { + bytes[offset + i] = value.charCodeAt(i); + } + }; + text(0, "RIFF"); + view.setUint32(4, bytes.length - 8, true); + text(8, "WAVEfmt "); + view.setUint32(16, 16, true); + view.setUint16(20, 1, true); + view.setUint16(22, 2, true); + view.setUint32(24, rate, true); + view.setUint32(28, rate * 4, true); + view.setUint16(32, 4, true); + view.setUint16(34, 16, true); + text(36, "data"); + view.setUint32(40, samples * 4, true); + let offset = 44; + for (const chunk of pcmChunks(packets, samplesToMs(samples, rate), rate)) { + for (const value of chunk) { + view.setInt16(offset, value, true); + offset += 2; + } + } + return { bytes, durationMs: samplesToMs(samples, rate) }; +} + +/** Render bounded stereo blocks. Resampling positions stay relative to each + * source packet, so a block boundary never resets interpolation phase. + * Blocks borrow a reusable buffer, valid until the iterator advances. */ +export function* pcmChunks( + packets: Packet[], + durationMs: number, + rate = CALL_SAMPLE_RATE, +): Generator { + const samples = Math.round(msToSamples(durationMs, rate)); + if ( + ![24000, 48000].includes(rate) || + !Number.isSafeInteger(samples) || + samples <= 0 || + samples > 1800 * rate + ) { + throw new Error("invalid_duration"); + } + const pending = packets + .map((packet, index) => { + if ( + ![0, 1].includes(packet.channel) || + ![1, 2].includes(packet.channels) || + !Number.isFinite(packet.rate) || + packet.rate < 8000 || + packet.rate > 96000 || + !Number.isFinite(packet.at) || + packet.at < 0 + ) { + throw new Error("invalid_packet"); + } + const start = Math.round(msToSamples(packet.at, rate)); + return { + packet, + index, + start, + end: + start + + Math.round( + (packet.pcm.length / packet.channels / packet.rate) * rate, + ), + }; + }) + .sort((a, b) => a.start - b.start || a.index - b.index); + let cursor = 0; + let active: typeof pending = []; + const buffer = new Int16Array((rate / 10) * 2); + for (let start = 0; start < samples; start += rate / 10) { + const end = Math.min(samples, start + rate / 10); + active = active.filter((p) => p.end > start); + while (cursor < pending.length && pending[cursor].start < end) { + active.push(pending[cursor++]); + } + // Overlapping packets retain the same last-writer behavior as WAV assembly. + active.sort((a, b) => a.index - b.index); + const chunk = buffer.subarray(0, (end - start) * 2); + chunk.fill(0); + for (const { packet: p, start: packetStart, end: packetEnd } of active) { + for ( + let i = Math.max(start, packetStart); + i < Math.min(end, packetEnd); + i++ + ) { + const position = ((i - packetStart) * p.rate) / rate, + lo = Math.floor(position), + alpha = position - lo; + let value = 0; + for (let channel = 0; channel < p.channels; channel++) { + const a = + p.pcm[Math.min(lo * p.channels + channel, p.pcm.length - 1)] ?? 0; + const b = + p.pcm[ + Math.min((lo + 1) * p.channels + channel, p.pcm.length - 1) + ] ?? a; + value += a + (b - a) * alpha; + } + chunk[(i - start) * 2 + p.channel] = Math.max( + -32768, + Math.min(32767, Math.round(value / p.channels)), + ); + } + } + yield chunk; + } +} diff --git a/integrations/audio/src/publisher.ts b/integrations/audio/src/publisher.ts new file mode 100644 index 000000000..561bdf990 --- /dev/null +++ b/integrations/audio/src/publisher.ts @@ -0,0 +1,171 @@ +export type RecordingSpan = { + spanId: string; + log: (event: { input?: unknown; metadata?: Record }) => void; +}; +import type { Segment } from "./segment"; +import type { AudioSegment } from "./segment"; +import { + mergeSelections, + segmentSelections, + type Selection, +} from "./selections"; + +export type TurnSelection = { + span: Pick; + intervals: Selection[]; + alias?: boolean; + metadata?: Record; +}; +export type RecordingContext = { + origin: number; + basis: string; + source: (channel: number) => Record; + closed: boolean; + metadata?: Record; +}; +type PublishedSegment = Pick< + AudioSegment, + "id" | "state" | "attachmentReference" +>; +type Snapshot = RecordingContext & { + segments: readonly Segment[]; + reason?: string; +}; +/** Publishes recording snapshots and uploaded attachment references. No framework types. */ +export class TracePublisher { + private publishedRecordings?: string; + private publishedSelections = new WeakMap(); + constructor( + private span: RecordingSpan, + private flush: () => Promise, + private snapshot: () => Snapshot, + private turnSelections: () => Iterable, + ) {} + async publish(segment?: PublishedSegment) { + // Retry telemetry once, independently of the completed export. + try { + await this.publishOnce(segment); + } catch { + await this.publishOnce(segment).catch(() => {}); + } + } + private async publishOnce(segment?: PublishedSegment) { + try { + // Retain only the reference so a logging failure can retry without uploading again. + if ( + segment?.state === "ready" && + segment.attachmentReference !== undefined + ) { + this.span.log({ + input: { audio: { [segment.id]: segment.attachmentReference } }, + }); + } + this.publishRecordings(); + this.publishSelections(); + await this.flush(); + } catch (error) { + // A failed flush may not have delivered any queued updates. Retry snapshots + // as well as the uploaded reference, without re-encoding or re-uploading. + this.publishedRecordings = undefined; + this.publishedSelections = new WeakMap(); + throw error; + } + } + publishRecordings() { + const snapshot = this.snapshot(); + const descriptors: Record[] = snapshot.segments.map( + (s) => ({ + id: s.id, + recording_group_id: "call", + state: s.state, + ...(s.reason ? { reason: s.reason } : {}), + ...(s.state === "ready" + ? { + attachment: { + span_id: this.span.spanId, + ref: `/input/audio/${s.id}`, + }, + mime_type: s.mimeType, + duration_ms: s.durationMs, + channel_count: 2, + sample_rate_hz: s.sampleRate, + } + : {}), + sources: s.channels.map(snapshot.source), + timeline: { + origin_unix_ms: snapshot.origin, + recording_start_offset_ms: s.start, + basis: snapshot.basis, + }, + ...(s.state === "omitted" + ? { + gaps: [ + { + start_offset_ms: s.start, + end_offset_ms: s.end, + reason: s.reason, + }, + ], + } + : {}), + }), + ); + if (snapshot.reason) { + descriptors.push({ + id: "call", + recording_group_id: "call", + state: "omitted", + reason: snapshot.reason, + truncated: true, + }); + } + if (!descriptors.length) { + descriptors.push({ + id: "call", + recording_group_id: "call", + state: snapshot.closed ? "omitted" : "pending", + ...(snapshot.closed ? { reason: "no_audio_observed" } : {}), + }); + } + // Compare the serialized wire value: callbacks can return freshly allocated or + // mutated metadata. Identity/version checks would miss those changes. + const signature = JSON.stringify({ + recordings: descriptors, + metadata: snapshot.metadata, + }); + if (this.publishedRecordings === signature) { + return; + } + this.span.log({ + metadata: { + "audio.recordings": descriptors, + ...snapshot.metadata, + }, + }); + this.publishedRecordings = signature; + } + publishSelections() { + const { segments } = this.snapshot(); + for (const turn of this.turnSelections()) { + const selections = segmentSelections( + mergeSelections(turn.intervals), + segments, + ); + const metadata = { + "audio.selections": selections, + ...(turn.alias + ? { + "audio.selection": selections.length === 1 ? selections[0] : null, + } + : {}), + ...turn.metadata, + }; + const signature = JSON.stringify(metadata); + if (this.publishedSelections.get(turn.span) === signature) { + continue; + } + turn.span.log({ metadata }); + this.publishedSelections.set(turn.span, signature); + } + } +} diff --git a/integrations/audio/src/recorder.test.ts b/integrations/audio/src/recorder.test.ts new file mode 100644 index 000000000..a462871a5 --- /dev/null +++ b/integrations/audio/src/recorder.test.ts @@ -0,0 +1,309 @@ +import type { AudioSegment } from "./segment"; +import { expect, test } from "vitest"; +import { Recorder, type RecordingOptions } from "./recorder"; +import { SegmentExporter, type Encoder } from "./exporter"; +import type { EncodedAudio } from "./segment"; +import { encodeCall } from "./pcm"; +const encode = async ( + packets: Parameters[0], + durationMs: number, +) => ({ + ...encodeCall(packets, durationMs), + mimeType: "audio/wav", + sampleRate: 24000, +}); +// Exercise the recorder with real segment execution and in-memory remote storage. +function recording( + options: RecordingOptions, + encode: Encoder, + publish: (s: AudioSegment) => Promise, +) { + let encoded: EncodedAudio; + const exporter = new SegmentExporter( + async (...args) => { + encoded = await encode(...args); + return encoded; + }, + () => ({ + reference: encoded, + upload: async () => ({ upload_status: "done" }), + }), + ); + return new Recorder(options, exporter, publish); +} +function accept(segment: AudioSegment) { + return segment.attachmentReference as EncodedAudio | undefined; +} +function samples(value: number, seconds = 1) { + return new Int16Array(24000 * seconds).fill(value); +} +test("exports progressively with bounded source memory across a ten minute timeline", async () => { + const files: { start: number; bytes: Uint8Array }[] = []; + const r = recording({ segmentDurationSeconds: 30 }, encode, async (s) => { + const data = accept(s); + if (data) { + files.push({ start: s.start, bytes: data.bytes }); + } + }); + for (let i = 0; i < 600; i++) { + r.record({ + pcm: samples(1000), + at: i * 1000, + rate: 24000, + channels: 1, + channel: 0, + }); + r.record({ + pcm: samples(-2000), + at: i * 1000, + rate: 24000, + channels: 1, + channel: 1, + }); + r.advance((i + 1) * 1000); + await r.drain(); + expect(r.retainedBytes).toBeLessThan(4 * 1024 * 1024); + } + expect(files.length).toBe(20); + await r.finish(); + expect(r.retainedBytes).toBe(0); + for (const [i, f] of files.entries()) { + expect(f.start).toBe(i * 30000); + const view = new DataView(f.bytes.buffer); + expect(view.getInt16(44, true)).toBe(1000); + expect(view.getInt16(f.bytes.length - 2, true)).toBe(-2000); + } +}); +test("splits a packet crossing the cut without losing its tail", async () => { + const files: Uint8Array[] = []; + const r = recording({ segmentDurationSeconds: 1 }, encode, async (s) => { + const data = accept(s); + if (data) { + files.push(data.bytes); + } + }); + r.record({ + pcm: samples(321, 3), + at: 0, + rate: 24000, + channels: 1, + channel: 0, + }); + r.advance(1000); + await r.drain(); + await r.finish(); + expect(files.map((f) => (f.length - 44) / 4)).toEqual([24000, 24000, 24000]); + expect(r.retainedBytes).toBe(0); +}); +test("rotated files reconstruct both source channels sample-for-sample including a partial final file", async () => { + const files: Uint8Array[] = []; + const recorder = recording( + { segmentDurationSeconds: 1 }, + encode, + async (s) => { + const data = accept(s); + if (data) { + files.push(data.bytes); + } + }, + ); + // Distinct nonconstant signals expose reordered, duplicated, or missing audio. + const length = 3 * 24000 + 1777; + const input = Int16Array.from( + { length }, + (_, i) => ((i * 17) % 30001) - 15000, + ); + const output = Int16Array.from( + { length }, + (_, i) => ((i * 31) % 28001) - 14000, + ); + // Packets deliberately cross the one-second file boundary. + for (let at = 0; at < length; at += 719) { + for (const [channel, source] of [input, output].entries()) { + recorder.record({ + pcm: source.slice(at, at + 719), + at: at / 24, + rate: 24000, + channels: 1, + channel: channel as 0 | 1, + }); + } + recorder.advance(Math.min(at + 719, length) / 24); + await recorder.drain(); + } + await recorder.finish(); + expect(files.map((file) => (file.length - 44) / 4)).toEqual([ + 24000, 24000, 24000, 1777, + ]); + const expected = Buffer.alloc(length * 4); + for (let i = 0; i < length; i++) { + expected.writeInt16LE(input[i], i * 4); + expected.writeInt16LE(output[i], i * 4 + 2); + } + expect( + Buffer.concat(files.map((file) => file.subarray(44))).equals(expected), + ).toBe(true); +}); + +test("hard duration limit preserves the prefix and reports truncation", async () => { + const r = recording({ maxDurationSeconds: 2 }, encode, async (s) => { + accept(s); + }); + for (let i = 0; i < 3; i++) { + r.record({ + pcm: samples(1), + at: i * 1000, + rate: 24000, + channels: 1, + channel: 0, + }); + } + await r.finish(); + expect(r.reason).toBe("duration_limit"); + expect(r.segments[0]).toMatchObject({ state: "ready", end: 2000 }); + expect(r.retainedBytes).toBe(0); +}); +test("slow publication bounds retained audio without erasing a completed segment", async () => { + let unblock!: () => void; + const blocked = new Promise((r) => { + unblock = r; + }); + const r = recording( + { segmentDurationSeconds: 1, maxBufferBytes: 100000 }, + encode, + async (s) => { + await blocked; + accept(s); + }, + ); + r.record({ pcm: samples(1), at: 0, rate: 24000, channels: 1, channel: 0 }); + r.advance(1000); + for (let i = 1; i < 8; i++) { + r.record({ + pcm: samples(1), + at: i * 1000, + rate: 24000, + channels: 1, + channel: 0, + }); + } + expect(r.retainedBytes).toBeLessThanOrEqual(100000); + expect(r.reason).toBe("capture_byte_limit"); + unblock(); + await r.finish(); + expect(r.segments.some((s) => s.state === "ready")).toBe(true); + expect(r.retainedBytes).toBe(0); +}); + +test.each(["encoding", "publication"])( + "%s failure retains earlier ready audio and releases source leases", + async (failure) => { + let count = 0, + encodings = 0; + const r = recording( + { segmentDurationSeconds: 1 }, + async (...args) => { + if (++encodings === 2 && failure === "encoding") { + throw new Error("encode failed"); + } + return encode(...args); + }, + async (s) => { + const data = accept(s); + if (data && ++count === 2 && failure === "publication") { + throw Error("export failed"); + } + }, + ); + for (let i = 0; i < 2; i++) { + r.record({ + pcm: samples(1), + at: i * 1000, + rate: 24000, + channels: 1, + channel: 0, + }); + r.advance((i + 1) * 1000); + await r.drain(); + } + await r.finish(); + expect(r.segments.map((s) => s.state)).toEqual([ + "ready", + failure === "encoding" ? "omitted" : "ready", + ]); + expect(r.retainedBytes).toBe(0); + }, +); + +test("a watermark behind the current origin cannot move a segment backwards", async () => { + const r = recording({ segmentDurationSeconds: 0.1 }, encode, async (s) => { + accept(s); + }); + r.record({ pcm: samples(1), at: 200, rate: 24000, channels: 1, channel: 0 }); + r.advance(500); + await r.drain(); + expect(r.segments.map((s) => [s.start, s.end])).toEqual([ + [200, 300], + [300, 400], + [400, 500], + ]); + r.advance(100); + await r.finish(); + expect(r.segments.map((s) => [s.start, s.end])).toEqual( + Array.from({ length: 10 }, (_, i) => [200 + i * 100, 300 + i * 100]), + ); +}); +test("soft byte pressure rotates before the duration threshold", async () => { + const r = recording( + { segmentDurationSeconds: 60, maxBufferBytes: 200000, flushFraction: 0.5 }, + encode, + async (s) => { + accept(s); + }, + ); + for (let i = 0; i < 3; i++) { + r.record({ + pcm: samples(1), + at: i * 1000, + rate: 24000, + channels: 1, + channel: 0, + }); + } + r.advance(2000); + await r.drain(); + expect(r.segments[0]).toMatchObject({ state: "ready", end: 2000 }); + await r.finish(); + expect(r.retainedBytes).toBe(0); +}); + +test.each([1, 60])( + "a delayed watermark drains %i-second clips before close without requiring more input", + async (seconds) => { + const durations: number[] = []; + const r = recording( + { segmentDurationSeconds: seconds }, + encode, + async (s) => { + const data = accept(s); + if (data) { + durations.push(data.durationMs); + } + }, + ); + for (let i = 0; i < 5 * seconds; i++) { + r.record({ + pcm: samples(123), + at: i * 1000, + rate: 24000, + channels: 1, + channel: 0, + }); + } + r.advance(5000 * seconds); + await r.drain(); + expect(durations).toEqual(Array(5).fill(1000 * seconds)); + await r.finish(); + expect(durations).toHaveLength(5); + }, +); diff --git a/integrations/audio/src/recorder.ts b/integrations/audio/src/recorder.ts new file mode 100644 index 000000000..93d8aabd7 --- /dev/null +++ b/integrations/audio/src/recorder.ts @@ -0,0 +1,288 @@ +import { msToSamples, samplesToMs, pcmBytesToMs } from "./timeline"; +import { AudioSegment, type Packet } from "./segment"; +export type { Packet } from "./segment"; +import { PcmBuffers } from "./buffers"; +import type { SegmentExporter } from "./exporter"; +export type RecordingOptions = { + segmentDurationSeconds?: number; + maxDurationSeconds?: number; + maxBufferBytes?: number; + flushFraction?: number; +}; +/** Framework-independent recorder. The adapter supplies a safe, monotonic cut watermark. */ +export class Recorder { + readonly options: Readonly>; + private packets: Packet[] = []; + readonly segments: AudioSegment[] = []; + readonly formats = new Map< + number, + Map + >(); + reason?: string; + private buffers: PcmBuffers; + get retainedBytes() { + return this.buffers.bytes; + } + private job?: Promise; + private segmentStartMs = 0; + private recordedEndMs = 0; + private earliestPacketMs = Infinity; + private closed = false; + private finishing?: Promise; + private confirmedThroughMs = 0; + constructor( + options: RecordingOptions, + private exporter: SegmentExporter, + private publish: (segment: AudioSegment) => Promise, + private pending: (segment: AudioSegment) => void = () => {}, + ) { + this.options = Object.freeze({ + segmentDurationSeconds: options.segmentDurationSeconds ?? 60, + maxDurationSeconds: options.maxDurationSeconds ?? 1800, + maxBufferBytes: options.maxBufferBytes ?? 32 * 1024 * 1024, + flushFraction: options.flushFraction ?? 0.5, + }); + validateOptions(this.options); + this.buffers = new PcmBuffers(this.options.maxBufferBytes); + } + + stop(reason: string) { + this.reason ??= reason; + } + copy(pcm: Int16Array): Int16Array | undefined { + if (this.closed || this.reason) { + return; + } + try { + return this.buffers.copy(pcm); + } catch (error) { + this.stop(error instanceof Error ? error.message : "capture_failed"); + return; + } + } + /** Discard staged audio; admitted audio belongs to the recorder/worker. */ + release(pcm: Int16Array) { + this.buffers.discard(pcm); + } + append(packet: Packet): boolean { + const duration = pcmBytesToMs( + packet.pcm.byteLength, + packet.rate, + packet.channels, + ); + const reason = this.validatePacket(packet, duration); + if (reason) { + this.stop(reason); + } + if (this.closed || this.reason) { + this.release(packet.pcm); + return false; + } + if (!this.buffers.admit(packet.pcm)) { + this.stop("invalid_pcm_ownership"); + return false; + } + const formats = this.formats.get(packet.channel) ?? new Map(); + formats.set(`${packet.rate}:${packet.channels}`, { + sample_rate_hz: packet.rate, + channel_count: packet.channels, + }); + this.formats.set(packet.channel, formats); + this.packets.push(packet); + this.earliestPacketMs = Math.min(this.earliestPacketMs, packet.at); + this.recordedEndMs = Math.max(this.recordedEndMs, packet.at + duration); + return true; + } + private validatePacket( + packet: Packet, + durationMs: number, + ): string | undefined { + if (!isSupportedPacket(packet)) { + return "unsupported_audio_format"; + } + if (packet.at < this.segmentStartMs - samplesToMs(1)) { + return "capture_clock_discontinuity"; + } + if (packet.at + durationMs > this.options.maxDurationSeconds * 1000) { + return "duration_limit"; + } + if (this.packets.length >= 100000) { + return "packet_limit"; + } + return undefined; + } + + record(packet: Packet): boolean { + const pcm = this.copy(packet.pcm); + return pcm ? this.append({ ...packet, pcm }) : false; + } + advance(watermark: number) { + if ( + this.closed || + this.reason || + !Number.isFinite(watermark) || + watermark <= this.segmentStartMs + ) { + return; + } + this.confirmedThroughMs = Math.max(this.confirmedThroughMs, watermark); + this.pump(); + } + // One job per recorder. Completion resumes eligible work without requiring + // another frame or blocking the application. Source limits bound the backlog. + private pump() { + if (this.job || !this.packets.length || (this.reason && !this.closed)) { + return; + } + const limit = this.closed + ? this.recordedEndMs + : Math.min(this.confirmedThroughMs, this.recordedEndMs); + const duration = this.options.segmentDurationSeconds * 1000; + if (this.earliestPacketMs >= this.segmentStartMs + duration) { + this.segmentStartMs += + Math.floor( + (Math.min(this.earliestPacketMs, limit) - this.segmentStartMs) / + duration, + ) * duration; + } + const cut = this.segmentStartMs + duration; + if (limit >= cut) { + this.rotate(cut); + } else if ( + this.closed || + this.retainedBytes >= + this.options.maxBufferBytes * this.options.flushFraction + ) { + this.rotate(limit); + } + } + private rotate(cut: number) { + if (!this.packets.length || cut <= this.segmentStartMs) { + return; + } + const { before, later } = this.splitPackets(cut); + if (!before.length) { + return; + } + const segment = new AudioSegment( + `call-${String(this.segments.length).padStart(4, "0")}`, + this.segmentStartMs, + cut, + before, + ); + this.packets = later; + this.earliestPacketMs = later.reduce((v, p) => Math.min(v, p.at), Infinity); + this.segmentStartMs = cut; + this.segments.push(segment); + try { + this.pending(segment); + } catch { + // Pending telemetry is best effort; upload publication sends the full recording descriptors. + } + this.job = this.exportSegment(segment).finally(() => { + this.job = undefined; + this.pump(); + }); + // Capture never awaits this task; drain()/finish() observe completion. + void this.job.catch(() => {}); + } + + private async exportSegment(segment: AudioSegment): Promise { + await this.exporter.export(segment, (pcm) => this.buffers.release(pcm)); + if (segment.state === "omitted") { + this.stop(segment.reason!); + } + // Trace publication cannot change the outcome of an audio export. + await this.publish(segment).catch(() => {}); + } + + /** Partition at a sample boundary; only a crossing packet requires copies. */ + private splitPackets(cutMs: number): { before: Packet[]; later: Packet[] } { + const before: Packet[] = [], + later: Packet[] = []; + for (const packet of this.packets) { + const count = Math.min( + packet.pcm.length, + Math.max( + 0, + Math.round(msToSamples(cutMs - packet.at, packet.rate)) * + packet.channels, + ), + ); + if (!count) { + later.push(packet); + } else if (count === packet.pcm.length) { + before.push({ ...packet, at: packet.at - this.segmentStartMs }); + } else { + // Splitting preserves the original reservation: the two new allocations + // have the same total source size and the old allocation is discarded. + const [head, tail] = this.buffers.split(packet.pcm, count); + before.push({ + ...packet, + pcm: head, + at: packet.at - this.segmentStartMs, + }); + later.push({ + ...packet, + pcm: tail, + at: packet.at + samplesToMs(count / packet.channels, packet.rate), + }); + } + } + return { before, later }; + } + + async drain() { + while (this.job) { + await this.job; + } + } + finish(): Promise { + if (!this.finishing) { + this.finishing = this.finishOnce(); + } + return this.finishing; + } + private async finishOnce() { + this.closed = true; + this.pump(); + await this.drain(); + // Unsubmitted copies (e.g. cancelled output) have no worker owner now. + this.buffers.close(); + this.packets = []; + } +} + +function validateOptions(options: Required): void { + const { + segmentDurationSeconds, + maxDurationSeconds, + maxBufferBytes, + flushFraction, + } = options; + const positiveLimits = [ + segmentDurationSeconds, + maxDurationSeconds, + maxBufferBytes, + ]; + if ( + !positiveLimits.every((value) => Number.isFinite(value) && value > 0) || + !Number.isInteger(maxBufferBytes) || + !(flushFraction > 0 && flushFraction < 1) + ) { + throw new Error("Invalid recording limits"); + } +} + +function isSupportedPacket(packet: Packet): boolean { + return ( + Number.isFinite(packet.at) && + packet.at >= 0 && + [0, 1].includes(packet.channel) && + [1, 2].includes(packet.channels) && + Number.isFinite(packet.rate) && + packet.rate >= 8000 && + packet.rate <= 96000 && + packet.pcm.length % packet.channels === 0 + ); +} diff --git a/integrations/audio/src/recording.test.ts b/integrations/audio/src/recording.test.ts new file mode 100644 index 000000000..fd04a8d01 --- /dev/null +++ b/integrations/audio/src/recording.test.ts @@ -0,0 +1,354 @@ +import { expect, test, vi } from "vitest"; +import { createRecording, type RecordingConfig, type Selection } from "./index"; + +test.each(["wav", "ogg"] as const)( + "public recording API exports %s only after upload succeeds", + async (audioFormat) => { + const log = vi.fn(); + const flush = vi.fn(async () => {}); + let complete!: () => void; + let started!: () => void; + const uploading = new Promise((resolve) => { + started = resolve; + }); + const uploaded = new Promise((resolve) => { + complete = resolve; + }); + const createAttachment = vi.fn( + (options: { + data: Uint8Array; + filename: string; + contentType: string; + }) => ({ + reference: { + key: "uploaded", + filename: options.filename, + content_type: options.contentType, + }, + upload: async () => { + started(); + await uploaded; + return { upload_status: "done" }; + }, + }), + ); + const recording = createRecording({ + audioFormat, + // Explicit undefined must behave like an omitted optional setting. + options: { + segmentDurationSeconds: undefined, + maxDurationSeconds: undefined, + maxBufferBytes: undefined, + flushFraction: undefined, + }, + span: { spanId: "call", log }, + flush, + createAttachment, + snapshot: () => ({ + origin: 1234, + basis: "test", + + source: (channel_index) => ({ channel_index }), + }), + turnSelections: () => [], + }); + recording.record({ + pcm: new Int16Array(480).fill(1000), + at: 0, + rate: 24000, + channels: 1, + channel: 0, + }); + const finishing = recording.finish(); + try { + await uploading; + recording.publishRecordings(); + expect( + log.mock.calls + .flatMap(([row]) => row.metadata?.["audio.recordings"] ?? []) + .every((r) => r.state === "pending"), + ).toBe(true); + expect(log.mock.calls.some(([row]) => row.input?.audio)).toBe(false); + const file = createAttachment.mock.calls[0][0]; + expect(Buffer.from(file.data.subarray(0, 4)).toString()).toBe( + audioFormat === "wav" ? "RIFF" : "OggS", + ); + expect(file.contentType).toBe(`audio/${audioFormat}`); + complete(); + await finishing; + expect( + log.mock.calls + .flatMap(([row]) => row.metadata?.["audio.recordings"] ?? []) + .at(-1), + ).toMatchObject({ state: "ready", mime_type: `audio/${audioFormat}` }); + expect(flush).toHaveBeenCalled(); + } finally { + complete(); + await finishing; + } + }, +); + +test.each(["pending", "attachment", "manifest", "selection", "flush"])( + "%s publication failure preserves uploaded audio and subsequent capture", + async (failure) => { + const rows: Parameters[0][] = []; + const audioKeys = (row: (typeof rows)[number]) => + row.input && typeof row.input === "object" && "audio" in row.input + ? Object.keys(row.input.audio as object) + : []; + let fail = true; + const rejectOnce = (stage: string) => { + if (stage === failure && fail) { + fail = false; + throw new Error("temporary trace failure"); + } + }; + const log = (row: (typeof rows)[number]) => { + if (audioKeys(row).length) { + rejectOnce("attachment"); + } + const manifest = row.metadata?.["audio.recordings"]; + if (Array.isArray(manifest)) { + const ready = manifest.some( + (s: { state: string }) => s.state === "ready", + ); + rejectOnce(ready ? "manifest" : "pending"); + } + if (row.metadata?.["audio.selections"]) { + rejectOnce("selection"); + } + rows.push(row); + }; + const turn = { log }; + const upload = vi.fn(async () => ({ upload_status: "done" })); + const recording = createRecording({ + audioFormat: "wav", + options: { segmentDurationSeconds: 0.02 }, + span: { spanId: "session", log }, + flush: async () => { + // A rejected flush need not have delivered the queued trace updates. + if (failure === "flush" && fail) { + rows.length = 0; + } + rejectOnce("flush"); + }, + createAttachment: ({ filename }) => ({ + reference: { key: filename }, + upload, + }), + snapshot: () => ({ + origin: 0, + basis: "test", + + source: (channel_index) => ({ channel_index }), + }), + turnSelections: () => [ + { + span: turn, + intervals: [ + { + recording_span_id: "session", + recording_id: "call", + channel_index: 0, + start_offset_ms: 0, + end_offset_ms: 40, + }, + ], + }, + ], + }); + try { + for (const at of [0, 20]) { + expect( + recording.record({ + pcm: new Int16Array(480), + at, + rate: 24000, + channels: 1, + channel: 0, + }), + ).toBe(true); + expect(() => recording.advance(at + 20)).not.toThrow(); + await recording.drain(); + expect(rows.flatMap(audioKeys)).toContain( + at === 0 ? "call-0000" : "call-0001", + ); + const selected = rows + .filter((r) => r.metadata?.["audio.selections"]) + .at(-1); + expect(selected?.metadata?.["audio.selections"]).toHaveLength( + at / 20 + 1, + ); + } + } finally { + await recording.finish(); + } + recording.publishRecordings(); + recording.publishSelections(); + expect(fail).toBe(false); + expect(upload).toHaveBeenCalledTimes(2); + const manifest = rows + .flatMap((r) => + Array.isArray(r.metadata?.["audio.recordings"]) + ? r.metadata["audio.recordings"] + : [], + ) + .slice(-2); + expect(manifest).toMatchObject([{ state: "ready" }, { state: "ready" }]); + expect(new Set(rows.flatMap(audioKeys))).toEqual( + new Set(["call-0000", "call-0001"]), + ); + expect(rows.at(-1)?.metadata?.["audio.selections"]).toMatchObject([ + { recording_id: "call-0000" }, + { recording_id: "call-0001" }, + ]); + }, +); + +test("staged audio is copied, can be discarded, and is bounded independently per recording", async () => { + const source = new Int16Array(12 * 1024 * 1024); + source[0] = 123; + const recordings = Array.from({ length: 3 }, () => + createRecording({ + options: { maxBufferBytes: source.byteLength }, + span: { spanId: "session", log: () => {} }, + flush: async () => {}, + createAttachment: () => { + throw new Error("Discarded audio must not upload"); + }, + snapshot: () => ({ + origin: 0, + basis: "test", + + source: () => ({}), + }), + turnSelections: () => [], + }), + ); + try { + // More than 64 MiB in aggregate must not trigger a hidden cross-session limit. + const staged = recordings.map((recording) => recording.copy(source)); + expect(staged.every((pcm) => pcm?.[0] === 123)).toBe(true); + source[0] = 456; + for (const [index, recording] of recordings.entries()) { + expect(staged[index]?.[0]).toBe(123); + recording.release(staged[index]!); + } + // Discarding returns capacity; exceeding this recording's own limit stops capture. + expect(recordings[0].copy(source)).toBeDefined(); + expect(recordings[0].copy(new Int16Array(1))).toBeUndefined(); + expect(recordings[0].reason).toBe("capture_byte_limit"); + expect(recordings[1].copy(new Int16Array(1))).toBeDefined(); + expect(recordings[1].reason).toBeUndefined(); + } finally { + await Promise.all(recordings.map((recording) => recording.finish())); + } +}); + +test("recording publication observes changed context and suppresses identical updates", async () => { + const log = vi.fn(); + const metadata = { source: "microphone" }; + const recording = createRecording({ + span: { spanId: "session", log }, + flush: async () => {}, + createAttachment: () => { + throw new Error("No audio captured"); + }, + snapshot: () => ({ + origin: 0, + basis: "test", + + source: () => ({}), + metadata, + }), + turnSelections: () => [], + }); + recording.publishRecordings(); + recording.publishRecordings(); + expect(log).toHaveBeenCalledTimes(1); + metadata.source = "replacement"; + recording.publishRecordings(); + expect(log).toHaveBeenCalledTimes(2); + expect(log.mock.calls.at(-1)?.[0].metadata.source).toBe("replacement"); + await recording.finish(); +}); + +test.each([undefined, "capture_byte_limit"])( + "finish publishes final state without audio (reason=%s) and is idempotent", + async (reason) => { + const log = vi.fn(); + const flush = vi.fn(async () => {}); + const recording = createRecording({ + span: { spanId: "session", log }, + flush, + createAttachment: () => { + throw new Error("No audio to upload"); + }, + snapshot: () => ({ origin: 0, basis: "test", source: () => ({}) }), + turnSelections: () => [], + }); + recording.publishRecordings(); + if (reason) { + recording.stop(reason); + } + await recording.finish(); + expect( + log.mock.calls.at(-1)?.[0].metadata["audio.recordings"], + ).toMatchObject([ + { state: "omitted", reason: reason ?? "no_audio_observed" }, + ]); + const count = log.mock.calls.length; + await recording.finish(); + expect(log).toHaveBeenCalledTimes(count); + expect(flush).toHaveBeenCalledTimes(1); + expect(recording.copy(new Int16Array(1))).toBeUndefined(); + }, +); + +test("finish publishes late turn selections without uploading the segment again", async () => { + const log = vi.fn(); + const turn = { log: vi.fn() }; + const upload = vi.fn(async () => ({ upload_status: "done" })); + const intervals: Selection[] = []; + const recording = createRecording({ + audioFormat: "wav", + options: { segmentDurationSeconds: 0.02 }, + span: { spanId: "session", log }, + flush: async () => {}, + createAttachment: () => ({ reference: { key: "audio" }, upload }), + snapshot: () => ({ + origin: 0, + basis: "test", + source: (channel_index) => ({ channel_index }), + }), + turnSelections: () => [{ span: turn, intervals }], + }); + recording.record({ + pcm: new Int16Array(480), + at: 0, + rate: 24000, + channels: 1, + channel: 0, + }); + recording.advance(20); + await recording.drain(); + intervals.push({ + recording_span_id: "session", + recording_id: "call", + channel_index: 0, + start_offset_ms: 0, + end_offset_ms: 20, + }); + await recording.finish(); + expect(turn.log.mock.calls.at(-1)?.[0].metadata["audio.selections"]).toEqual([ + { + recording_span_id: "session", + recording_id: "call-0000", + channel_index: 0, + start_offset_ms: 0, + end_offset_ms: 20, + }, + ]); + expect(upload).toHaveBeenCalledTimes(1); +}); diff --git a/integrations/audio/src/recording.ts b/integrations/audio/src/recording.ts new file mode 100644 index 000000000..5862eeabf --- /dev/null +++ b/integrations/audio/src/recording.ts @@ -0,0 +1,112 @@ +import { Recorder, type RecordingOptions } from "./recorder"; +import { + TracePublisher, + type RecordingSpan, + type RecordingContext, + type TurnSelection, +} from "./publisher"; +import { SegmentExporter, type CreateAttachment } from "./exporter"; +import { createEncoder } from "./worker"; +import { oggOpus } from "./codec"; +import type { Packet } from "./segment"; + +export interface RecordingConfig { + options?: RecordingOptions; + audioFormat?: "ogg" | "wav"; + encoder?: { module: string }; + span: RecordingSpan; + flush: () => Promise; + snapshot: () => Omit; + /** Latest turn spans and call-relative audio intervals; used to publish audio.selections. */ + turnSelections: () => Iterable; + createAttachment: CreateAttachment; +} +/** One recording session. Pipeline components remain private to the package. */ +class Recording { + private readonly recorder: Recorder; + private readonly publisher: TracePublisher; + private closed = false; + private finishing?: Promise; + + constructor(config: RecordingConfig) { + this.recorder = new Recorder( + config.options ?? {}, + new SegmentExporter( + createEncoder( + config.encoder ?? + (config.audioFormat === "wav" ? undefined : oggOpus()), + ), + config.createAttachment, + ), + (segment) => this.publisher.publish(segment), + () => this.publisher.publishRecordings(), + ); + this.publisher = new TracePublisher( + config.span, + config.flush, + () => ({ + ...config.snapshot(), + closed: this.closed, + segments: this.recorder.segments, + reason: this.recorder.reason, + }), + config.turnSelections, + ); + } + + get reason() { + return this.recorder.reason; + } + get formats(): ReadonlyMap< + number, + ReadonlyMap + > { + return this.recorder.formats; + } + /** Copy a timestamped PCM frame into the recording. */ + record(packet: Packet): boolean { + return this.recorder.record(packet); + } + /** Confirm that no more frames will arrive before this recording offset. */ + advance(confirmedThroughMs: number): void { + this.recorder.advance(confirmedThroughMs); + } + stop(reason: string): void { + this.recorder.stop(reason); + } + drain(): Promise { + return this.recorder.drain(); + } + /** Finish the final segment and wait for encoding, uploads, and trace publication. */ + finish(): Promise { + if (!this.finishing) { + this.closed = true; + this.finishing = this.recorder + .finish() + .then(() => this.publisher.publish()); + } + return this.finishing; + } + + // Staged playback: copy now, then append only the audio actually played, or release it. + copy(pcm: Int16Array): Int16Array | undefined { + return this.recorder.copy(pcm); + } + append(packet: Packet): boolean { + return this.recorder.append(packet); + } + release(pcm: Int16Array): void { + this.recorder.release(pcm); + } + publishRecordings(): void { + this.publisher.publishRecordings(); + } + publishSelections(): void { + this.publisher.publishSelections(); + } +} + +/** Create a framework-independent recording session. */ +export function createRecording(config: RecordingConfig) { + return new Recording(config); +} diff --git a/integrations/audio/src/segment.test.ts b/integrations/audio/src/segment.test.ts new file mode 100644 index 000000000..44106c311 --- /dev/null +++ b/integrations/audio/src/segment.test.ts @@ -0,0 +1,55 @@ +import { expect, test } from "vitest"; +import { AudioSegment } from "./segment"; + +const encoded = { + durationMs: 20, + sampleRate: 24000, + mimeType: "audio/wav", +}; +function segment() { + return new AudioSegment("call-0000", 40, 60, [ + { pcm: new Int16Array(480), at: 0, rate: 24000, channels: 1, channel: 0 }, + ]); +} +test("segment becomes ready only after encoding and upload, retaining its descriptor", () => { + const s = segment(); + expect(() => s.dispatch({ type: "ready", reference: "uploaded" })).toThrow(); + const packets = s.dispatch({ type: "encode" }); + expect(packets[0].pcm).toHaveLength(480); + expect(() => s.dispatch({ type: "encode" })).toThrow(); + s.dispatch({ type: "encoded", format: encoded }); + expect(s.state).toBe("pending"); + s.dispatch({ type: "ready", reference: "uploaded" }); + expect(s).toMatchObject({ + id: "call-0000", + start: 40, + end: 60, + state: "ready", + channels: [0], + ...encoded, + attachmentReference: "uploaded", + }); + expect(() => s.dispatch({ type: "encoded", format: encoded })).toThrow(); + expect(() => + s.dispatch({ type: "omitted", reason: "late_failure" }), + ).toThrow(); + expect(s.state).toBe("ready"); +}); +test.each(["queued", "encoding", "uploading"])( + "omission during %s is terminal and preserves its original reason", + (phase) => { + const s = segment(); + if (phase !== "queued") { + s.dispatch({ type: "encode" }); + } + if (phase === "uploading") { + s.dispatch({ type: "encoded", format: encoded }); + } + s.dispatch({ type: "omitted", reason: "failed" }); + expect(s).toMatchObject({ state: "omitted", reason: "failed" }); + expect(() => s.dispatch({ type: "ready", reference: "late" })).toThrow(); + expect(() => s.dispatch({ type: "encoded", format: encoded })).toThrow(); + expect(() => s.dispatch({ type: "omitted", reason: "other" })).toThrow(); + expect(s.reason).toBe("failed"); + }, +); diff --git a/integrations/audio/src/segment.ts b/integrations/audio/src/segment.ts new file mode 100644 index 000000000..d0a980463 --- /dev/null +++ b/integrations/audio/src/segment.ts @@ -0,0 +1,118 @@ +export type Packet = { + pcm: Int16Array; + rate: number; + channels: number; + channel: number; + at: number; +}; +export type EncodedAudio = { + bytes: Uint8Array; + durationMs: number; + mimeType: string; + sampleRate: number; +}; +export type Segment = Readonly<{ + id: string; + start: number; + end: number; + state: "pending" | "ready" | "omitted"; + reason?: string; + mimeType?: string; + sampleRate?: number; + durationMs?: number; + channels: readonly number[]; +}>; + +type AudioFormat = Omit; +type SegmentEvent = + | { type: "encode" } + | { type: "encoded"; format: AudioFormat } + | { type: "ready"; reference: unknown } + | { type: "omitted"; reason: string }; +type SegmentState = + | { phase: "queued"; packets: Packet[] } + | { phase: "encoding" } + | { phase: "uploading"; format: AudioFormat } + | { phase: "ready"; format: AudioFormat; reference: unknown } + | { phase: "omitted"; reason: string }; + +/** Owns segment transitions and its descriptor; the exporter owns encoded bytes. */ +export class AudioSegment implements Segment { + private current: SegmentState; + readonly channels: readonly number[]; + + constructor( + readonly id: string, + readonly start: number, + readonly end: number, + packets: Packet[], + ) { + this.current = { phase: "queued", packets }; + this.channels = [...new Set(packets.map((packet) => packet.channel))]; + } + + get state(): Segment["state"] { + const { phase } = this.current; + return phase === "ready" || phase === "omitted" ? phase : "pending"; + } + get reason() { + return this.current.phase === "omitted" ? this.current.reason : undefined; + } + private get format() { + return "format" in this.current ? this.current.format : undefined; + } + get mimeType() { + return this.format?.mimeType; + } + get sampleRate() { + return this.format?.sampleRate; + } + get durationMs() { + return this.format?.durationMs; + } + get attachmentReference() { + return this.current.phase === "ready" ? this.current.reference : undefined; + } + + dispatch(event: { type: "encode" }): Packet[]; + dispatch(event: Exclude): void; + dispatch(event: SegmentEvent): Packet[] | void { + const state = this.current; + switch (event.type) { + case "encode": + if (state.phase === "queued") { + this.current = { phase: "encoding" }; + return state.packets; + } + break; + case "encoded": + if (state.phase === "encoding") { + const { durationMs, mimeType, sampleRate } = event.format; + this.current = { + phase: "uploading", + format: { durationMs, mimeType, sampleRate }, + }; + return; + } + break; + case "ready": + if (state.phase === "uploading") { + this.current = { + phase: "ready", + format: state.format, + reference: event.reference, + }; + return; + } + break; + case "omitted": + if (state.phase !== "ready" && state.phase !== "omitted") { + this.current = { phase: "omitted", reason: event.reason }; + return; + } + } + throw new Error( + `Invalid audio segment transition: ${state.phase} -> ${event.type}`, + ); + } +} diff --git a/integrations/audio/src/selections.test.ts b/integrations/audio/src/selections.test.ts new file mode 100644 index 000000000..f4d8beb4f --- /dev/null +++ b/integrations/audio/src/selections.test.ts @@ -0,0 +1,68 @@ +import { expect, test } from "vitest"; +import { + mergeSelections, + segmentSelections, + type Selection, +} from "./selections"; +const range = (start: number, end: number): Selection => ({ + recording_span_id: "root", + recording_id: "call", + channel_index: 1, + start_offset_ms: start, + end_offset_ms: end, +}); +test("compaction preserves recording identities, channels and real discontinuities", () => { + const result = mergeSelections([ + range(20, 40), + range(0, 20), + range(50, 60), + { ...range(40, 50), recording_id: "other" }, + { ...range(40, 50), recording_span_id: "another-root" }, + { ...range(40, 50), channel_index: 0 }, + ]); + expect(result).toHaveLength(5); + expect(result).toContainEqual(range(0, 40)); + expect(result).toContainEqual(range(50, 60)); +}); +test("thirty minutes of contiguous frames scale with files, not frames", () => { + const minutes = 30; + const ranges = Array.from({ length: minutes * 60 * 50 }, (_, i) => + range(i * 20, (i + 1) * 20), + ); + const merged = mergeSelections(ranges); + expect(merged).toEqual([range(0, minutes * 60000)]); + const segments = Array.from({ length: minutes }, (_, i) => ({ + id: `call-${i}`, + start: i * 60000, + end: (i + 1) * 60000, + state: "ready" as const, + channels: [1], + })); + expect(segmentSelections(merged, segments)).toHaveLength(minutes); +}); + +test("pending and failed segments never produce selections; a crossing turn uses local file coordinates", () => { + const selections = segmentSelections( + [ + { + recording_span_id: "root", + recording_id: "call", + channel_index: 0, + start_offset_ms: 29000, + end_offset_ms: 32000, + }, + ], + [ + { id: "first", start: 0, end: 30000, state: "ready", channels: [0] }, + { id: "second", start: 30000, end: 60000, state: "ready", channels: [0] }, + { id: "pending", start: 0, end: 60000, state: "pending", channels: [0] }, + { id: "failed", start: 0, end: 60000, state: "omitted", channels: [0] }, + ], + ); + expect( + selections.map((s) => [s.recording_id, s.start_offset_ms, s.end_offset_ms]), + ).toEqual([ + ["first", 29000, 30000], + ["second", 0, 2000], + ]); +}); diff --git a/integrations/audio/src/selections.ts b/integrations/audio/src/selections.ts new file mode 100644 index 000000000..538be0fb7 --- /dev/null +++ b/integrations/audio/src/selections.ts @@ -0,0 +1,64 @@ +import { msToSamples, samplesToMs } from "./timeline"; +import type { Segment } from "./segment"; +export type Selection = { + recording_span_id: string; + recording_id: string; + start_offset_ms: number; + end_offset_ms: number; + channel_index: number; +}; +export function segmentSelections( + intervals: Selection[], + segments: readonly Segment[], +): Selection[] { + return segments + .filter((s) => s.state === "ready") + .flatMap((segment) => + intervals + .map((s) => ({ + ...s, + recording_id: segment.id, + start_offset_ms: Math.max( + 0, + samplesToMs( + Math.round(msToSamples(s.start_offset_ms - segment.start)), + ), + ), + end_offset_ms: Math.min( + segment.end - segment.start, + samplesToMs( + Math.round(msToSamples(s.end_offset_ms - segment.start)), + ), + ), + })) + .filter((s) => s.end_offset_ms > s.start_offset_ms), + ); +} + +export function mergeSelections(ranges: Selection[]): Selection[] { + const out: Selection[] = []; + for (const r of [...ranges].sort( + (a, b) => + a.recording_span_id.localeCompare(b.recording_span_id) || + a.recording_id.localeCompare(b.recording_id) || + a.channel_index - b.channel_index || + a.start_offset_ms - b.start_offset_ms, + )) { + const previous = out.at(-1); + if ( + previous && + previous.recording_span_id === r.recording_span_id && + previous.recording_id === r.recording_id && + previous.channel_index === r.channel_index && + previous.end_offset_ms >= r.start_offset_ms - samplesToMs(1) + ) { + previous.end_offset_ms = Math.max( + previous.end_offset_ms, + r.end_offset_ms, + ); + } else { + out.push({ ...r }); + } + } + return out; +} diff --git a/integrations/audio/src/timeline.ts b/integrations/audio/src/timeline.ts new file mode 100644 index 000000000..cc1937b93 --- /dev/null +++ b/integrations/audio/src/timeline.ts @@ -0,0 +1,23 @@ +/** Shared call-selection clock; source and encoded audio may use other rates. */ +export const CALL_SAMPLE_RATE = 24000; + +/** Fractional sample frames. Round only where an integer sample boundary is needed. */ +export function msToSamples( + milliseconds: number, + sampleRate = CALL_SAMPLE_RATE, +) { + return milliseconds * (sampleRate / 1000); +} + +export function samplesToMs(samples: number, sampleRate = CALL_SAMPLE_RATE) { + return samples / (sampleRate / 1000); +} + +/** Duration of interleaved signed 16-bit PCM. */ +export function pcmBytesToMs( + bytes: number, + sampleRate: number, + channels: number, +) { + return samplesToMs(bytes / 2 / channels, sampleRate); +} diff --git a/integrations/audio/src/worker.test.ts b/integrations/audio/src/worker.test.ts new file mode 100644 index 000000000..73fcb4e7e --- /dev/null +++ b/integrations/audio/src/worker.test.ts @@ -0,0 +1,53 @@ +import { expect, test } from "vitest"; +import { createEncoder } from "./worker"; +test("worker admission permits one running and one queued job, without blocking a caller", async () => { + const encoder = createEncoder(); + const packet = () => [ + { + pcm: new Int16Array(24000).fill(1234), + rate: 24000, + channels: 1, + channel: 0, + at: 0, + }, + ]; + const results = await Promise.allSettled([ + encoder(packet(), 1000), + encoder(packet(), 1000), + encoder(packet(), 1000), + ]); + expect(results.map((r) => r.status)).toEqual([ + "fulfilled", + "fulfilled", + "rejected", + ]); + const first = results[0]; + if (first.status === "fulfilled") { + expect(first.value.bytes.length).toBe(96044); + } + expect(results[2]).toMatchObject({ + reason: { message: "recording_worker_capacity" }, + }); +}); + +test("a failed codec releases admission for subsequent recordings", async () => { + await expect( + createEncoder({ module: "file:///missing-braintrust-test-codec.mjs" })( + [], + 20, + ), + ).rejects.toThrow(); + const result = await createEncoder()([], 20); + expect(result.mimeType).toBe("audio/wav"); + expect(result.bytes.length).toBe(1964); +}); + +test("successive segments reuse the codec runtime but keep independent output buffers", async () => { + const encoder = createEncoder({ + module: new URL("./fixtures/counting-codec.mjs", import.meta.url).href, + }); + const first = await encoder([], 20); + const second = await encoder([], 20); + expect([...first.bytes]).toEqual([1]); + expect([...second.bytes]).toEqual([2]); +}); diff --git a/integrations/audio/src/worker.ts b/integrations/audio/src/worker.ts new file mode 100644 index 000000000..37b7e799c --- /dev/null +++ b/integrations/audio/src/worker.ts @@ -0,0 +1,125 @@ +import { Worker } from "node:worker_threads"; +import { createRequire } from "node:module"; +import { dirname, join } from "node:path"; +import type { Encoder } from "./exporter"; +import type { EncodedAudio } from "./segment"; +const key = Symbol.for("braintrust.audio.workers.v1"); +type Pool = { + running: boolean; + queue: (() => void)[]; + worker?: Worker; + idle?: ReturnType; +}; +const pool: Pool = Reflect.get(globalThis, key) ?? { + running: false, + queue: [], +}; +Reflect.set(globalThis, key, pool); +async function acquire() { + if (pool.running) { + if (pool.queue.length >= 1) { + throw new Error("recording_worker_capacity"); + } + await new Promise((resolve) => pool.queue.push(resolve)); + } else { + pool.running = true; + } + clearTimeout(pool.idle); + pool.idle = undefined; +} +function release() { + const next = pool.queue.shift(); + if (next) { + next(); + } else { + pool.running = false; + const worker = pool.worker; + if (!worker) { + return; + } + // Reuse nearby work, but do not pin a codec isolate indefinitely after a call. + worker.unref(); + pool.idle = setTimeout(() => { + pool.idle = undefined; + if (pool.worker === worker && !pool.running) { + pool.worker = undefined; + void worker.terminate().catch(() => {}); + } + }, 10000); + pool.idle.unref(); + } +} +function getWorker() { + if (!pool.worker) { + const worker = new Worker( + join( + dirname( + createRequire(__filename).resolve("@braintrust/audio/package.json"), + ), + "dist/audio-worker.mjs", + ), + // Keep short-lived conversion objects from growing a large nursery in + // the codec isolate. This does not constrain the application's JS heap. + { + env: {}, + execArgv: [], + resourceLimits: { maxYoungGenerationSizeMb: 2 }, + }, + ); + pool.worker = worker; + const clear = () => { + if (pool.worker === worker) { + pool.worker = undefined; + } + }; + // Idle worker failures must neither become unhandled errors nor poison reuse. + worker.on("error", clear); + worker.on("exit", clear); + } + pool.worker.ref(); + return pool.worker; +} +export function createEncoder(codec?: { module: string }): Encoder { + return async (packets, durationMs) => { + await acquire(); + let worker: Worker | undefined; + let timer: ReturnType | undefined; + let cleanup = () => {}; + try { + worker = getWorker(); + return await new Promise((resolve, reject) => { + const message = (result: EncodedAudio & { reason?: string }) => + result.reason ? reject(new Error(result.reason)) : resolve(result); + const error = () => reject(new Error("encoding_failed")); + const exit = () => reject(new Error("encoder_exit")); + worker!.once("message", message); + worker!.once("error", error); + worker!.once("exit", exit); + cleanup = () => { + worker!.off("message", message); + worker!.off("error", error); + worker!.off("exit", exit); + }; + timer = setTimeout(() => reject(new Error("encoder_deadline")), 10000); + const buffers = [...new Set(packets.map((p) => p.pcm.buffer))].filter( + (b): b is ArrayBuffer => b instanceof ArrayBuffer, + ); + worker!.postMessage( + { packets, durationMs, encoderModule: codec?.module }, + buffers, + ); + }); + } catch (error) { + if (pool.worker === worker) { + pool.worker = undefined; + } + // Never reuse a failed/timed-out codec job; keep admission until it stops. + await worker?.terminate().catch(() => {}); + throw error; + } finally { + clearTimeout(timer); + cleanup(); + release(); + } + }; +} diff --git a/integrations/audio/tsconfig.json b/integrations/audio/tsconfig.json new file mode 100644 index 000000000..ec45517e8 --- /dev/null +++ b/integrations/audio/tsconfig.json @@ -0,0 +1,15 @@ +{ + "compilerOptions": { + "declaration": true, + "lib": ["es2022"], + "module": "es2022", + "target": "es2022", + "moduleResolution": "node", + "strict": true, + "noUnusedLocals": true, + "esModuleInterop": true, + "skipLibCheck": true + }, + "include": ["."], + "exclude": ["node_modules/**", "**/dist/**"] +} diff --git a/integrations/audio/tsup.config.ts b/integrations/audio/tsup.config.ts new file mode 100644 index 000000000..1b6f41e9c --- /dev/null +++ b/integrations/audio/tsup.config.ts @@ -0,0 +1,13 @@ +import { defineConfig } from "tsup"; +export default defineConfig({ + entry: { + index: "src/index.ts", + encoder: "src/encoder.ts", + "audio-worker": "src/audio-worker.ts", + }, + format: ["cjs", "esm"], + dts: true, + shims: true, + clean: true, + external: ["@audio/encode-opus"], +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 042bae613..adf5e522b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -127,6 +127,28 @@ importers: specifier: 4.3.6 version: 4.3.6 + integrations/audio: + dependencies: + '@audio/encode-opus': + specifier: 1.3.1 + version: 1.3.1 + devDependencies: + '@types/node': + specifier: ^20.10.5 + version: 20.19.40 + ogg-opus-decoder: + specifier: ^1.7.3 + version: 1.7.5 + tsup: + specifier: ^8.5.1 + version: 8.5.1(@swc/core@1.15.8)(jiti@2.6.1)(postcss@8.5.26)(tsx@4.21.0)(typescript@5.9.3)(yaml@2.9.0) + typescript: + specifier: ^5.3.3 + version: 5.9.3 + vitest: + specifier: 4.1.11 + version: 4.1.11(@opentelemetry/api@1.9.0)(@types/node@20.19.40)(msw@2.13.6(@types/node@20.19.40)(typescript@5.9.3))(vite@8.1.5(@types/node@20.19.40)(esbuild@0.28.1)(jiti@2.6.1)(terser@5.44.1)(tsx@4.21.0)(yaml@2.9.0)) + integrations/browser-js: dependencies: als-browser: @@ -627,6 +649,10 @@ packages: zod: optional: true + '@audio/encode-opus@1.3.1': + resolution: {integrity: sha512-FFNQgim/doIajnOyDidaIx541HI51VoMKWQk0NXFf2XP4bLgduR7kSmwoPXzF5TPCt7axC+Wu6IuTOGyuVWQPg==} + engines: {node: '>=18'} + '@babel/runtime@7.28.4': resolution: {integrity: sha512-Q/N6JNWvIvPnLDvjlE1OUBLPQHH6l3CltCEsHIujp45zQUSSh8K+gHnaEX45yAT1nyngnINhvWtzN+Nb9D8RAQ==} engines: {node: '>=6.9.0'} @@ -1036,6 +1062,9 @@ packages: cpu: [x64] os: [win32] + '@eshaz/web-worker@1.2.2': + resolution: {integrity: sha512-WxXiHFmD9u/owrzempiDlBB1ZYqiLnm9s6aPc8AlFQalq2tKmqdmMr9GXOupDgzXtqnBipj8Un0gkIm7Sjf8mw==} + '@eslint-community/eslint-utils@4.9.1': resolution: {integrity: sha512-phrYmNiYppR7znFEdqgfWHXR6NCkZEK7hwWDHZUjit/2/U0r6XvkDl0SYnoM51Hq7FhCGdLDT6zxCCOY1hexsQ==} engines: {node: ^12.22.0 || ^14.17.0 || >=16.0.0} @@ -2149,6 +2178,12 @@ packages: '@vitest/utils@4.1.11': resolution: {integrity: sha512-zTCVGpyFsGWBhllOyKlTw/vnr6D9qxsfSDyfbyZmTyjHw5N/VuvzHpHoQjm2ZJzn4RJgx5w4r7V0er69CmLgPQ==} + '@wasm-audio-decoders/common@9.0.7': + resolution: {integrity: sha512-WRaUuWSKV7pkttBygml/a6dIEpatq2nnZGFIoPTc5yPLkxL6Wk4YaslPM98OPQvWacvNZ+Py9xROGDtrFBDzag==} + + '@wasm-audio-decoders/opus-ml@0.0.3': + resolution: {integrity: sha512-SEeTmk0eb3q5etJA/NHCRWW6BivTx9tqvNTmG0kg2ynuATEQRUZnwQE6Ibvl1BBvqIGHqI57+qyJMWmJBwb5XQ==} + '@webassemblyjs/ast@1.14.1': resolution: {integrity: sha512-nuBEDgQfm1ccRp/8bCQrx1frohyufl4JlbMMZ4P1wpeOfDhF6FQkxZJ1b/e+PLwr6X1Nhw6OLme5usuBWYBvuQ==} @@ -2461,6 +2496,9 @@ packages: resolution: {integrity: sha512-BSeNnyus75C4//NQ9gQt1/csTXyo/8Sb+afLAkzAptFuMsod9HFokGNudZpi/oQV73hnVK+sR+5PVRMd+Dr7YQ==} engines: {node: '>=12'} + codec-parser@2.5.0: + resolution: {integrity: sha512-Ru9t80fV8B0ZiixQl8xhMTLru+dzuis/KQld32/x5T/+3LwZb0/YvQdSKytX9JqCnRdiupvAvyYJINKrXieziQ==} + color-convert@2.0.1: resolution: {integrity: sha512-RRECPsj7iu/xb5oKYcsFHSppFNnsj/52OVTRKb4zP5onXwVF3zVmmToNcOfGC+CRDpfK/U584fMg38ZHCaElKQ==} engines: {node: '>=7.0.0'} @@ -3516,6 +3554,9 @@ packages: resolution: {integrity: sha512-XrsrhT5sybtKI6wakr2SPOlGZWWYbUXZ7a0jT8/QOeAPau+1X/bSegNe5YR75oJmEZQbKningirmGOEJCIk61Q==} engines: {node: '>=12.20.0'} + ogg-opus-decoder@1.7.5: + resolution: {integrity: sha512-08WfRbuBBew1H63PqQ3zj2B/J0Eh9ISfNOTi+Dk6vuXmkBGSx8BiPJGOJ470EYeKezTa7/GhOAwHe3XlXwOZMw==} + on-finished@2.4.1: resolution: {integrity: sha512-oVlzkg3ENAhCk2zdv7IJwd/QUD4z2RxRwpkcGY8psCVcCYZNq4wYnVWALHM+brtuJjePWiYF/ClmuDr8Ch5+kg==} engines: {node: '>= 0.8'} @@ -3567,6 +3608,9 @@ packages: resolution: {integrity: sha512-6IpQ7mKUxRcZNLIObR0hz7lxsapSSIYNZJwXPGeF0mTVqGKFIXj1DQcMoT22S3ROcLyY/rz0PWaWZ9ayWmad9g==} engines: {node: '>= 0.8.0'} + opus-decoder@0.7.12: + resolution: {integrity: sha512-6zpHFDi/MhYvMwyYiuJPqJ2lcRkfm7wHblSvT3lUtx/kLvHkunoe/rhTn1dLKxqVGLeQj3gQi7CnR0N2p3RuOg==} + outdent@0.5.0: resolution: {integrity: sha512-/jHxFIzoMXdqPzTaCpFzAAWhpkSjZPF4Vsn6jAfNpmbH/ymsmd7Qc6VE9BGn0L6YMj6uwpQLxCECpus4ukKS9Q==} @@ -3915,6 +3959,9 @@ packages: resolution: {integrity: sha512-bzyZ1e88w9O1iNJbKnOlvYTrWPDl46O1bG0D3XInv+9tkPrxrN8jUUTiFlDkkmKWgn1M6CfIA13SuGqOa9Korw==} engines: {node: '>=14'} + simple-yenc@1.0.4: + resolution: {integrity: sha512-5gvxpSd79e9a3V4QDYUqnqxeD4HGlhCakVpb6gMnDD7lexJggSBJRBO5h52y/iJrdXRilX9UCuDaIJhSWm5OWw==} + sisteransi@1.0.5: resolution: {integrity: sha512-bLGGlR1QxBcynn2d5YmDX4MGjlZvy2MRBDRNHLJ8VI6l6+9FUiyTFNJ0IveOSP0bcXgVDPRcfGqA0pjaqUpfVg==} @@ -4572,6 +4619,8 @@ snapshots: optionalDependencies: zod: 3.25.76 + '@audio/encode-opus@1.3.1': {} + '@babel/runtime@7.28.4': {} '@braintrust/bt-darwin-arm64@0.12.0': @@ -4972,6 +5021,8 @@ snapshots: '@esbuild/win32-x64@0.28.1': optional: true + '@eshaz/web-worker@1.2.2': {} + '@eslint-community/eslint-utils@4.9.1(eslint@9.39.4(jiti@2.6.1))': dependencies: eslint: 9.39.4(jiti@2.6.1) @@ -6147,6 +6198,15 @@ snapshots: convert-source-map: 2.0.0 tinyrainbow: 3.1.1 + '@wasm-audio-decoders/common@9.0.7': + dependencies: + '@eshaz/web-worker': 1.2.2 + simple-yenc: 1.0.4 + + '@wasm-audio-decoders/opus-ml@0.0.3': + dependencies: + '@wasm-audio-decoders/common': 9.0.7 + '@webassemblyjs/ast@1.14.1': dependencies: '@webassemblyjs/helper-numbers': 1.13.2 @@ -6460,6 +6520,8 @@ snapshots: strip-ansi: 6.0.1 wrap-ansi: 7.0.0 + codec-parser@2.5.0: {} + color-convert@2.0.1: dependencies: color-name: 1.1.4 @@ -7566,6 +7628,13 @@ snapshots: obug@2.2.1: {} + ogg-opus-decoder@1.7.5: + dependencies: + '@wasm-audio-decoders/common': 9.0.7 + '@wasm-audio-decoders/opus-ml': 0.0.3 + codec-parser: 2.5.0 + opus-decoder: 0.7.12 + on-finished@2.4.1: dependencies: ee-first: 1.1.1 @@ -7603,6 +7672,10 @@ snapshots: type-check: 0.4.0 word-wrap: 1.2.5 + opus-decoder@0.7.12: + dependencies: + '@wasm-audio-decoders/common': 9.0.7 + outdent@0.5.0: {} outvariant@1.4.3: {} @@ -7996,6 +8069,8 @@ snapshots: signal-exit@4.1.0: {} + simple-yenc@1.0.4: {} + sisteransi@1.0.5: {} slash@3.0.0: {}