refactor: extracted audio and voice engine into a standalone library crate

- Extracted audio capture and playback implementations, along with the WebRTC peer engine, from `pi-natives` into a new `pi-voice` library crate.
- Updated `pi-natives` bindings to consume the extracted `pi_voice` audio streams and live peer core.
- Added release validation gate jobs, parallelized Linux binary builds, and introduced a concurrent macOS release build job in the CI workflow.
- Updated Bazel workspace configurations, Cargo manifests, and documentation to include the new `pi-voice` crate and its dependencies.

.bazelrc

@@ -22,7 +22,8 @@ build --@rules_rust//rust/toolchain/channel=nightly
 
 # --- Rust validation (replaces cargo clippy/nextest in CI) --------------------
 # Mirrors cargo semantics: crates with `[lints] workspace = true` (pi-ast,
-# pi-iso, pi-natives, pi-shell, pi-walker) get the strict workspace policy
+# pi-iso, pi-natives, pi-shell, pi-voice, pi-walker) get the strict workspace
+# policy
 # (clippy-strict; bazel/clippy.bazelrc is generated from Cargo.toml); everything
 # else gets default clippy + -Dwarnings (clippy).
 build:clippy --aspects=@rules_rust//rust:defs.bzl%rust_clippy_aspect

.github/workflows/ci.yml

@@ -75,7 +75,10 @@ jobs:
    # ref (or from a tagged main HEAD) is also treated as a release.
    release_metadata:
       name: Resolve release metadata
-      runs-on: ${{ github.event_name == 'pull_request' && 'ubuntu-22.04' || 'omp-kata' }}
+      # Hosted, not omp-kata: this ~20 s git-tag probe gates the whole release
+      # chain, and queueing it behind the 4-runner kata pool adds dead minutes
+      # to the darwin critical path.
+      runs-on: ubuntu-22.04
       outputs:
          is-release: ${{ steps.detect.outputs.is-release }}
          release-tag: ${{ steps.detect.outputs.release-tag }}
@@ -152,11 +155,11 @@ jobs:
          # brush fork is exempt (same as run-rs-task.ts's cargo excludes).
          - name: Clippy (workspace lint policy on opted-in crates)
            run: |
-              bazelisk query "kind('rust_library|rust_shared_library', //crates/pi-ast/... + //crates/pi-iso/... + //crates/pi-natives/... + //crates/pi-shell/... + //crates/pi-walker/...)" \
+              bazelisk query "kind('rust_library|rust_shared_library', //crates/pi-ast/... + //crates/pi-iso/... + //crates/pi-natives/... + //crates/pi-shell/... + //crates/pi-voice/... + //crates/pi-walker/...)" \
                 | xargs bazelisk --bazelrc="${{ steps.cache.outputs.rc }}" build --config=clippy-strict --
          - name: Clippy (default lints elsewhere)
            run: |
-              bazelisk query "kind('rust_library|rust_shared_library', //crates/... - (//crates/pi-ast/... + //crates/pi-iso/... + //crates/pi-natives/... + //crates/pi-shell/... + //crates/pi-walker/...) - //crates/vendor/brush-core/... - //crates/vendor/brush-builtins/...)" \
+              bazelisk query "kind('rust_library|rust_shared_library', //crates/... - (//crates/pi-ast/... + //crates/pi-iso/... + //crates/pi-natives/... + //crates/pi-shell/... + //crates/pi-voice/... + //crates/pi-walker/...) - //crates/vendor/brush-core/... - //crates/vendor/brush-builtins/...)" \
                 | xargs bazelisk --bazelrc="${{ steps.cache.outputs.rc }}" build --config=clippy --
          - name: Rustfmt
            run: bazelisk --bazelrc="${{ steps.cache.outputs.rc }}" build --config=rustfmt //crates/...
@@ -401,8 +404,11 @@ jobs:
               OMP_INSTALL_TEST_SKIP_NATIVE_BUILD: "1"
            run: bun run ci:test:install-methods
 
-   release_binary:
+   # Aggregates every validation job so publish-side release jobs gate on one
-      name: "Release binary: ${{ matrix.target_id }}"
+   # result. Binary BUILDS deliberately do not wait for this gate — they run
+   # in parallel with the test fan-out and only publishing is held back.
+   release_gate:
+      name: Release validation gate
       if: ${{ needs.release_metadata.outputs.is-release == 'true' && !cancelled() &&
          needs.rust_validate.result == 'success' &&
          needs.native_addons.result == 'success' &&
@@ -415,6 +421,20 @@ jobs:
          needs.test_smoke.result == 'success' && needs.check.result == 'success' &&
          needs.install_methods.result == 'success' }}
       needs: [release_metadata, check, rust_validate, native_addons, test_workspace, test_coding_agent_singleton, test_ts_native, test_coding_agent_ui, test_coding_agent_runtime, test_coding_agent_native, test_smoke, install_methods]
+      runs-on: ubuntu-22.04
+      steps:
+         - run: echo "release validation green"
+
+   # Builds (does not publish) the Linux-hosted release binaries in parallel
+   # with the test fan-out; the single need is native_addons, whose artifact
+   # supplies their addons. Darwin builds live in release_binary_darwin so
+   # they start at release_metadata time instead. Publishing (npm leaves,
+   # GitHub release, core npm) is gated on release_gate downstream.
+   release_binary:
+      name: "Release binary: ${{ matrix.target_id }}"
+      if: ${{ needs.release_metadata.outputs.is-release == 'true' && !cancelled() &&
+         needs.native_addons.result == 'success' }}
+      needs: [release_metadata, native_addons]
       strategy:
          fail-fast: false
          matrix:
@@ -453,22 +473,6 @@ jobs:
                     binary_path: packages/coding-agent/binaries/omp-linux-musl-arm64,
                     native_targets: linux-musl-arm64,
                  }
-               - {
-                    os: macos-15-intel,
-                    platform: darwin,
-                    arch: x64,
-                    target_id: darwin-x64,
-                    binary_path: packages/coding-agent/binaries/omp-darwin-x64,
-                    native_targets: darwin-x64-baseline,
-                 }
-               - {
-                    os: macos-14,
-                    platform: darwin,
-                    arch: arm64,
-                    target_id: darwin-arm64,
-                    binary_path: packages/coding-agent/binaries/omp-darwin-arm64,
-                    native_targets: darwin-arm64,
-                 }
                - {
                     os: ubuntu-22.04,
                     platform: win32,
@@ -480,60 +484,25 @@ jobs:
       runs-on: ${{ matrix.os }}
       permissions:
          contents: read
-         id-token: write
-      env:
-         MACOS_SIGNING: ${{ secrets.APPLE_CERTIFICATE_P12 != '' && secrets.APPLE_CERTIFICATE_PASSWORD != '' && secrets.APPLE_API_KEY_ID != '' && secrets.APPLE_API_ISSUER_ID != '' && secrets.APPLE_API_KEY != '' }}
       steps:
          - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
          - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2.2.0
            with:
               bun-version: "1.3"
-         - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0
-           with:
-              node-version: "24"
-              registry-url: "https://registry.npmjs.org"
-         # npm runs under Bun when invoked by the release script; npm 12
-         # requires a newer emulated Node version than Bun 1.3 provides.
-         - name: Ensure npm supports trusted publishing
-           if: ${{ !inputs.skip_npm }}
-           run: npm install -g npm@11.17.0
          - name: Cache bun dependencies
            uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
            with:
               path: ~/.bun/install/cache
               key: bun-${{ runner.os }}-${{ hashFiles('**/bun.lock') }}
          - run: bun install --frozen-lockfile
-         # Linux and Windows addons come from the native_addons job. Darwin runners
-         # build only their own architecture because cross-hosted artifacts do
-         # not exist for macOS.
          - name: Install prebuilt native addon(s)
-           if: matrix.platform != 'darwin'
            uses: ./.github/actions/native-artifacts
            with:
               targets: ${{ matrix.native_targets }}
-         - name: Build native addon (bazel)
-           if: matrix.platform == 'darwin'
-           uses: ./.github/actions/bazel-natives
-           with:
-              targets: ${{ matrix.native_targets }}
-              cache-scope: release-${{ matrix.target_id }}
          - name: Build release binary
            env:
               RELEASE_TARGETS: ${{ matrix.target_id }}
            run: bun run ci:release:build-binaries
-         - name: Sign and notarize macOS binary (Developer ID)
-           # Replaces the ad-hoc signature with a Developer ID + hardened-runtime
-           # one (+JIT/library-validation entitlements; omp dlopens its
-           # runtime-extracted native addon, which has a different Team ID) and
-           # notarizes. Auto-skips until the APPLE_* secrets are configured.
-           if: matrix.platform == 'darwin' && env.MACOS_SIGNING == 'true'
-           env:
-              APPLE_CERTIFICATE_P12: ${{ secrets.APPLE_CERTIFICATE_P12 }}
-              APPLE_CERTIFICATE_PASSWORD: ${{ secrets.APPLE_CERTIFICATE_PASSWORD }}
-              APPLE_API_KEY_ID: ${{ secrets.APPLE_API_KEY_ID }}
-              APPLE_API_ISSUER_ID: ${{ secrets.APPLE_API_ISSUER_ID }}
-              APPLE_API_KEY: ${{ secrets.APPLE_API_KEY }}
-           run: bash scripts/ci-macos-sign.sh "${{ matrix.binary_path }}"
          # Windows binary is cross-built on Linux, so we have no Windows runner
          # to smoke it on. Cross-build correctness is verified via the bun
          # `--compile --target=bun-windows-x64-*` cross-compile. Musl binaries
@@ -557,25 +526,168 @@ jobs:
                 HOME="$runtime_dir/home" XDG_DATA_HOME="$runtime_dir/xdg" omp --version
                 HOME="$runtime_dir/home" XDG_DATA_HOME="$runtime_dir/xdg" omp --smoke-test
               '
-         - name: Publish native addon package
-           if: ${{ !inputs.skip_npm && matrix.libc != 'musl' }}
-           env:
-              # Fallback auth: setup-node wrote an .npmrc referencing
-              # NODE_AUTH_TOKEN; npm uses it only when OIDC has no trusted
-              # publisher for the package (or on a first publish).
-              NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
-           run: bun run ci:release:publish-native-leaf ${{ matrix.target_id }}
          - name: Upload release binary artifact
            uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
            with:
               name: omp-binary-${{ matrix.target_id }}
               path: ${{ matrix.binary_path }}
 
+   # Darwin release builds are the longest leg of a release (bazel-built
+   # addons on macOS runners), build their own addons, and touch nothing from
+   # native_addons — so they gate on release_metadata alone and start the
+   # moment the release run is detected, fully overlapped with the tests.
+   release_binary_darwin:
+      name: "Release binary: ${{ matrix.target_id }}"
+      if: ${{ needs.release_metadata.outputs.is-release == 'true' && !cancelled() }}
+      needs: [release_metadata]
+      strategy:
+         fail-fast: false
+         matrix:
+            include:
+               - {
+                    os: macos-15-intel,
+                    target_id: darwin-x64,
+                    binary_path: packages/coding-agent/binaries/omp-darwin-x64,
+                    native_targets: darwin-x64-baseline,
+                 }
+               - {
+                    os: macos-14,
+                    target_id: darwin-arm64,
+                    binary_path: packages/coding-agent/binaries/omp-darwin-arm64,
+                    native_targets: darwin-arm64,
+                 }
+      runs-on: ${{ matrix.os }}
+      permissions:
+         contents: read
+      env:
+         MACOS_SIGNING: ${{ secrets.APPLE_CERTIFICATE_P12 != '' && secrets.APPLE_CERTIFICATE_PASSWORD != '' && secrets.APPLE_API_KEY_ID != '' && secrets.APPLE_API_ISSUER_ID != '' && secrets.APPLE_API_KEY != '' }}
+      steps:
+         - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+         - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2.2.0
+           with:
+              bun-version: "1.3"
+         - name: Cache bun dependencies
+           uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
+           with:
+              path: ~/.bun/install/cache
+              key: bun-${{ runner.os }}-${{ hashFiles('**/bun.lock') }}
+         - run: bun install --frozen-lockfile
+         # Darwin runners build their own architecture; cross-hosted artifacts
+         # do not exist for macOS. The warm workflow keeps these scopes seeded
+         # near HEAD, so this is normally the version-bump delta.
+         - name: Build native addon (bazel)
+           uses: ./.github/actions/bazel-natives
+           with:
+              targets: ${{ matrix.native_targets }}
+              cache-scope: release-${{ matrix.target_id }}
+         - name: Build release binary
+           env:
+              RELEASE_TARGETS: ${{ matrix.target_id }}
+           run: bun run ci:release:build-binaries
+         - name: Sign and notarize macOS binary (Developer ID)
+           # Replaces the ad-hoc signature with a Developer ID + hardened-runtime
+           # one (+JIT/library-validation entitlements; omp dlopens its
+           # runtime-extracted native addon, which has a different Team ID) and
+           # notarizes. Auto-skips until the APPLE_* secrets are configured.
+           if: env.MACOS_SIGNING == 'true'
+           env:
+              APPLE_CERTIFICATE_P12: ${{ secrets.APPLE_CERTIFICATE_P12 }}
+              APPLE_CERTIFICATE_PASSWORD: ${{ secrets.APPLE_CERTIFICATE_PASSWORD }}
+              APPLE_API_KEY_ID: ${{ secrets.APPLE_API_KEY_ID }}
+              APPLE_API_ISSUER_ID: ${{ secrets.APPLE_API_ISSUER_ID }}
+              APPLE_API_KEY: ${{ secrets.APPLE_API_KEY }}
+           run: bash scripts/ci-macos-sign.sh "${{ matrix.binary_path }}"
+         - name: Smoke release binary
+           run: |
+              runtime_dir="$(mktemp -d)"
+              HOME="$runtime_dir/home" XDG_DATA_HOME="$runtime_dir/xdg" "${{ matrix.binary_path }}" --version
+              HOME="$runtime_dir/home" XDG_DATA_HOME="$runtime_dir/xdg" "${{ matrix.binary_path }}" --smoke-test
+         # Darwin addons exist only on these runners; export them so
+         # release_native_leaves can publish every leaf package after the
+         # validation gate. Non-darwin leaves reuse the native-addons artifact.
+         - name: Upload darwin native addon artifact
+           uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+           with:
+              name: native-addons-${{ matrix.target_id }}
+              path: packages/natives/native/pi_natives.${{ matrix.target_id }}*.node
+              if-no-files-found: error
+              retention-days: 1
+         - name: Upload release binary artifact
+           uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+           with:
+              name: omp-binary-${{ matrix.target_id }}
+              path: ${{ matrix.binary_path }}
+
+   # Publishes the five @oh-my-pi/pi-natives-<tag> leaf packages once
+   # validation passes and every binary built. Runs on one linux runner: leaf
+   # publishing is platform-agnostic file packaging (gen-npm-packages reads
+   # packages/natives/native), so no per-platform runners are needed.
+   release_native_leaves:
+      name: Publish native leaf packages
+      if: ${{ needs.release_metadata.outputs.is-release == 'true' && !cancelled() &&
+         needs.release_gate.result == 'success' &&
+         needs.release_binary.result == 'success' &&
+         needs.release_binary_darwin.result == 'success' &&
+         !inputs.skip_npm }}
+      needs: [release_metadata, release_gate, release_binary, release_binary_darwin]
+      runs-on: ubuntu-22.04
+      # id-token lets npm mint the OIDC token for trusted publishing; the
+      # NODE_AUTH_TOKEN below is the fallback for unconfigured packages.
+      permissions:
+         id-token: write
+         contents: read
+      steps:
+         - uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 # v6.0.3
+         - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2.2.0
+           with:
+              bun-version: "1.3"
+         - uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0
+           with:
+              node-version: "24"
+              registry-url: "https://registry.npmjs.org"
+         # npm runs under Bun when invoked by the release script; npm 12
+         # requires a newer emulated Node version than Bun 1.3 provides.
+         - name: Ensure npm supports trusted publishing
+           run: npm install -g npm@11.17.0
+         - name: Cache bun dependencies
+           uses: actions/cache@0057852bfaa89a56745cba8c7296529d2fc39830 # v4.3.0
+           with:
+              path: ~/.bun/install/cache
+              key: bun-${{ runner.os }}-${{ hashFiles('**/bun.lock') }}
+         - run: bun install --frozen-lockfile
+         # Linux + win32 addons from the native_addons artifact (gnu targets
+         # only — musl reuses the linux filenames and has no leaf package).
+         - name: Install linux and win32 native addons
+           uses: ./.github/actions/native-artifacts
+           with:
+              targets: linux-x64-baseline linux-x64-modern linux-arm64 win32-x64-baseline
+         # Darwin addons come flat (canonical filenames) from the
+         # release_binary_darwin legs.
+         - name: Install darwin native addons
+           uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+           with:
+              pattern: native-addons-darwin-*
+              path: packages/natives/native
+              merge-multiple: true
+         - name: Publish native leaf packages
+           env:
+              # Fallback auth: setup-node wrote an .npmrc referencing
+              # NODE_AUTH_TOKEN; npm uses it only when OIDC has no trusted
+              # publisher for the package (or on a first publish).
+              NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
+           run: |
+              set -euo pipefail
+              for tag in linux-x64 linux-arm64 darwin-x64 darwin-arm64 win32-x64; do
+                bun run ci:release:publish-native-leaf "$tag"
+              done
+
    release_github:
       name: Publish GitHub release
       if: ${{ needs.release_metadata.outputs.is-release == 'true' && !cancelled() &&
-         needs.release_binary.result == 'success' }}
+         needs.release_gate.result == 'success' &&
-      needs: [release_metadata, release_binary]
+         needs.release_binary.result == 'success' &&
+         needs.release_binary_darwin.result == 'success' }}
+      needs: [release_metadata, release_gate, release_binary, release_binary_darwin]
       runs-on: ubuntu-22.04
       permissions:
          contents: write
@@ -646,9 +758,14 @@ jobs:
       name: Publish to npm
       if: ${{ needs.release_metadata.outputs.is-release == 'true' && !cancelled() &&
          needs.release_binary.result == 'success' &&
+         needs.release_binary_darwin.result == 'success' &&
          needs.release_github_verify.result == 'success' &&
+         needs.release_native_leaves.result == 'success' &&
          !inputs.skip_npm }}
-      needs: [release_metadata, release_binary, release_github_verify]
+      # release_native_leaves must finish first: the core package lists the
+      # leaf packages as lockstep-versioned optionalDependencies, so installs
+      # resolve them the moment the core version is live.
+      needs: [release_metadata, release_binary, release_binary_darwin, release_github_verify, release_native_leaves]
       runs-on: ubuntu-22.04
       # `id-token: write` lets npm mint the GitHub OIDC token it exchanges for a
       # short-lived publish token (trusted publishing + provenance). When a

Cargo.lock

@@ -4818,9 +4818,7 @@ dependencies = [
  "anyhow",
  "arboard",
  "ast-grep-core",
- "audiopus_sys",
  "base64",
- "bytes",
  "clap",
  "clipboard-win",
  "core-graphics",
@@ -4838,17 +4836,16 @@ dependencies = [
  "image",
  "inferno",
  "libc",
- "maudio",
  "napi",
  "napi-build",
  "napi-derive",
- "opus",
  "parking_lot",
  "phf 0.13.1",
  "pi-ast",
  "pi-iso",
  "pi-shell",
  "pi-uutils-ctx",
+ "pi-voice",
  "pi-walker",
  "png",
  "portable-pty",
@@ -4865,7 +4862,6 @@ dependencies = [
  "toml",
  "unicode-segmentation",
  "unicode-width 0.2.2",
- "webrtc",
  "windows-sys 0.61.2",
  "winreg 0.56.0",
  "x11rb",
@@ -4961,6 +4957,20 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "pi-voice"
+version = "17.1.8"
+dependencies = [
+ "audiopus_sys",
+ "bytes",
+ "flume",
+ "maudio",
+ "opus",
+ "parking_lot",
+ "tokio",
+ "webrtc",
+]
+
 [[package]]
 name = "pi-walker"
 version = "17.1.8"

Cargo.toml

@@ -177,6 +177,7 @@ too_many_lines = "allow"     # Arbitrary limits don't account for necessary comp
 pi-ast = { path = "crates/pi-ast" }
 pi-iso = { path = "crates/pi-iso" }
 pi-shell = { path = "crates/pi-shell" }
+pi-voice = { path = "crates/pi-voice" }
 pi-walker = { path = "crates/pi-walker" }
 brush-core = { path = "crates/vendor/brush-core" }
 brush-builtins = { path = "crates/vendor/brush-builtins" }

crates/pi-natives/BUILD.bazel

@@ -48,6 +48,7 @@ rust_shared_library(
         "//crates/pi-iso",
         "//crates/pi-shell",
         "//crates/pi-uutils-ctx",
+        "//crates/pi-voice",
         "//crates/pi-walker",
     ],
 )

crates/pi-natives/Cargo.toml

@@ -15,11 +15,9 @@ workspace = true
 
 [dependencies]
 anyhow.workspace = true
-audiopus_sys.workspace = true
 arboard.workspace = true
 ast-grep-core.workspace = true
 base64.workspace = true
-bytes.workspace = true
 clap.workspace = true
 globset.workspace = true
 fontdue.workspace = true
@@ -32,16 +30,15 @@ icy_sixel.workspace = true
 ignore.workspace = true
 image = { workspace = true, features = ["bmp"] }
 inferno.workspace = true
-maudio.workspace = true
 napi.workspace = true
 napi-derive.workspace = true
-opus.workspace = true
 parking_lot.workspace = true
 phf.workspace = true
 flume.workspace = true
 pi-ast.workspace = true
 pi-iso.workspace = true
 pi-shell.workspace = true
+pi-voice.workspace = true
 pi-walker.workspace = true
 pi-uutils-ctx = { path = "../pi-uutils-ctx" }
 portable-pty.workspace = true
@@ -58,7 +55,6 @@ tokio-util.workspace = true
 toml.workspace = true
 unicode-segmentation.workspace = true
 unicode-width.workspace = true
-webrtc.workspace = true
 xxhash-rust.workspace = true
 
 [target.'cfg(target_os = "linux")'.dependencies]
@@ -72,16 +68,6 @@ xcap = { version = "=0.9.6", default-features = false }
 [target.'cfg(target_os = "macos")'.dependencies]
 tempfile = "=3.27.0"
 core-graphics = "0.25"
-# Regenerate miniaudio FFI bindings from the compiled headers on macOS only.
-# The maudio-sys pregenerated `unix.rs` bindings are Linux-shaped: their
-# `ma_device`/`ma_context` backend-state unions omit the `#ifdef
-# MA_SUPPORT_COREAUDIO` members that the macOS build of miniaudio.c actually
-# contains, so shipping them on darwin is an ABI mismatch that silently breaks
-# CoreAudio capture (device init/start succeed but no render callbacks fire).
-# darwin builds with host Xcode (libclang available), so bindgen is fine here;
-# the hermetic zig/musl/windows toolchains keep the bindgen-free pregen path,
-# whose bindings are correct for those platforms.
-maudio = { workspace = true, features = ["generate-bindings"] }
 
 [target.'cfg(unix)'.dependencies]
 libc.workspace = true

crates/pi-natives/src/audio.rs

@@ -1,310 +1,26 @@
-//! Cross-platform microphone capture and streaming speaker playback.
+//! N-API bindings for microphone capture and speaker playback.
 //!
-//! miniaudio owns platform device discovery, format conversion, channel mixing,
+//! The engine — device discovery, format conversion, mixing, drain semantics —
-//! and resampling. The N-API classes expose one stable mono `f32` contract to
+//! lives in `pi_voice::audio`; these classes adapt its mono `f32` contract to
-//! TypeScript while the internal playback stream is shared with native WebRTC.
+//! TypeScript callbacks and `Float32Array` buffers.
 
-use std::sync::{
+use std::sync::Arc;
-	Arc,
-	atomic::{AtomicBool, AtomicU32, Ordering},
-};
 
-use flume::TryRecvError;
-use maudio::{
-	audio::{performance::PerformanceProfile, sample_rate::SampleRate},
-	backend::Backend,
-	device::{
-		Device,
-		device_builder::{DeviceBuilder, DeviceBuilderOps},
-	},
-};
 use napi::{
 	bindgen_prelude::{Float32Array, Result},
 	threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode, UnknownReturnValue},
 };
 use napi_derive::napi;
 use parking_lot::Mutex;
-use tokio::sync::Notify;
+use pi_voice::audio::{CaptureStream, PlaybackState, PlaybackStream};
-
-const AUDIO_CHANNELS: u32 = 1;
-// PulseAudio TCP playback stutters with a 20 ms target buffer; 50 ms absorbs
-// transport jitter while preserving interactive latency.
-#[cfg(target_os = "linux")]
-const PLAYBACK_PERIOD_MS: u32 = 50;
-#[cfg(not(target_os = "linux"))]
-const PLAYBACK_PERIOD_MS: u32 = 20;
-// miniaudio's PulseAudio backend reserves three periods. Android's OpenSL ES
-// source emits 125 ms fragments, so Linux capture needs at least 150 ms queued.
-#[cfg(target_os = "linux")]
-const CAPTURE_PERIOD_MS: u32 = 50;
-#[cfg(not(target_os = "linux"))]
-const CAPTURE_PERIOD_MS: u32 = 20;
-// PulseAudio can retain its default three periods after the producer closes.
-// Wait for all of them before stopping the device so the tail reaches the sink.
-#[cfg(target_os = "linux")]
-const PLAYBACK_DRAIN_CALLBACKS: usize = 3;
-#[cfg(not(target_os = "linux"))]
-const PLAYBACK_DRAIN_CALLBACKS: usize = 2;
-
-#[cfg(target_os = "macos")]
-const AUDIO_BACKENDS: &[Backend] = &[Backend::CoreAudio];
-#[cfg(target_os = "windows")]
-const AUDIO_BACKENDS: &[Backend] = &[Backend::Wasapi];
-#[cfg(target_os = "linux")]
-const AUDIO_BACKENDS: &[Backend] = &[Backend::PulseAudio, Backend::Alsa, Backend::Jack];
-#[cfg(not(any(target_os = "macos", target_os = "windows", target_os = "linux")))]
-const AUDIO_BACKENDS: &[Backend] = &[Backend::Sndio, Backend::Audio4, Backend::Oss];
 
 type CaptureCallback = ThreadsafeFunction<Float32Array, UnknownReturnValue>;
-type NativeResult<T> = std::result::Result<T, String>;
-
-struct PlaybackState {
-	gain_bits: AtomicU32,
-	drained:   AtomicBool,
-	stopped:   AtomicBool,
-	notify:    Notify,
-}
-
-impl PlaybackState {
-	fn new() -> Self {
-		Self {
-			gain_bits: AtomicU32::new(1.0f32.to_bits()),
-			drained:   AtomicBool::new(false),
-			stopped:   AtomicBool::new(false),
-			notify:    Notify::new(),
-		}
-	}
-
-	fn gain(&self) -> f32 {
-		f32::from_bits(self.gain_bits.load(Ordering::Acquire))
-	}
-
-	fn set_gain(&self, gain: f32) {
-		self.gain_bits.store(gain.to_bits(), Ordering::Release);
-	}
-
-	fn mark_drained(&self) {
-		if !self.drained.swap(true, Ordering::AcqRel) {
-			self.notify.notify_waiters();
-		}
-	}
-
-	fn mark_stopped(&self) {
-		self.stopped.store(true, Ordering::Release);
-		self.notify.notify_waiters();
-	}
-
-	async fn wait_for_drain(&self) {
-		loop {
-			let notified = self.notify.notified();
-			if self.drained.load(Ordering::Acquire) || self.stopped.load(Ordering::Acquire) {
-				return;
-			}
-			notified.await;
-		}
-	}
-}
-
-/// Producer endpoint for one native playback device.
-#[derive(Clone)]
-pub(crate) struct PlaybackWriter {
-	tx:    flume::Sender<Vec<f32>>,
-	state: Arc<PlaybackState>,
-}
-
-impl PlaybackWriter {
-	/// Queue mono floating-point samples without blocking the caller.
-	pub(crate) fn write(&self, samples: &[f32]) -> NativeResult<()> {
-		if samples.is_empty() {
-			return Ok(());
-		}
-		if self.state.stopped.load(Ordering::Acquire) || self.state.drained.load(Ordering::Acquire) {
-			return Err("Native audio playback is closed".to_owned());
-		}
-		self
-			.tx
-			.send(samples.to_vec())
-			.map_err(|_| "Native audio playback is closed".to_owned())
-	}
-}
-
-/// Running mono playback stream shared by N-API playback and native WebRTC.
-pub(crate) struct PlaybackStream {
-	device: Option<Device<f32>>,
-	writer: Option<PlaybackWriter>,
-	state:  Arc<PlaybackState>,
-}
-
-impl PlaybackStream {
-	/// Open and start the default speaker at the requested logical sample rate.
-	pub(crate) fn start(sample_rate: u32) -> NativeResult<Self> {
-		let sample_rate = audio_sample_rate(sample_rate)?;
-		let state = Arc::new(PlaybackState::new());
-		let (tx, rx) = flume::unbounded::<Vec<f32>>();
-		let callback_state = Arc::clone(&state);
-		let mut current = Vec::new();
-		let mut cursor = 0;
-		let mut empty_callbacks = 0;
-		let mut builder = DeviceBuilder::playback().f32();
-		builder
-			.sample_rate(sample_rate)
-			.playback_channels(AUDIO_CHANNELS)
-			.period_size_millis(PLAYBACK_PERIOD_MS)
-			.performance_profile(PerformanceProfile::LowLatency)
-			.backends(AUDIO_BACKENDS);
-		let mut device = builder
-			.with_callback(move |_device, output| {
-				fill_playback(
-					&rx,
-					&mut current,
-					&mut cursor,
-					output,
-					&callback_state,
-					&mut empty_callbacks,
-				);
-			})
-			.map_err(|error| format!("Failed to open the default speaker: {error}"))?;
-		device
-			.device_start()
-			.map_err(|error| format!("Failed to start speaker playback: {error}"))?;
-
-		Ok(Self {
-			device: Some(device),
-			writer: Some(PlaybackWriter { tx, state: Arc::clone(&state) }),
-			state,
-		})
-	}
-
-	/// Clone the producer endpoint used by the remote-audio decoder.
-	pub(crate) fn writer(&self) -> NativeResult<PlaybackWriter> {
-		self
-			.writer
-			.clone()
-			.ok_or_else(|| "Native audio playback is closed".to_owned())
-	}
-
-	fn state(&self) -> Arc<PlaybackState> {
-		Arc::clone(&self.state)
-	}
-
-	fn finish_input(&mut self) {
-		self.writer.take();
-	}
-
-	fn set_gain(&self, gain: f32) -> NativeResult<()> {
-		if !gain.is_finite() {
-			return Err("Audio playback gain must be finite".to_owned());
-		}
-		self.state.set_gain(gain.max(0.0));
-		Ok(())
-	}
-
-	/// Stop playback immediately and release the default speaker.
-	pub(crate) fn stop(&mut self) -> NativeResult<()> {
-		self.writer.take();
-		self.state.mark_stopped();
-		let Some(mut device) = self.device.take() else {
-			return Ok(());
-		};
-		device
-			.device_stop()
-			.map_err(|error| format!("Failed to stop speaker playback: {error}"))
-	}
-}
-
-impl Drop for PlaybackStream {
-	fn drop(&mut self) {
-		let _ = self.stop();
-	}
-}
-
-fn audio_sample_rate(sample_rate: u32) -> NativeResult<SampleRate> {
-	SampleRate::try_from(sample_rate)
-		.map_err(|error| format!("Unsupported audio sample rate {sample_rate}: {error}"))
-}
-
-fn fill_playback(
-	rx: &flume::Receiver<Vec<f32>>,
-	current: &mut Vec<f32>,
-	cursor: &mut usize,
-	output: &mut [f32],
-	state: &PlaybackState,
-	empty_callbacks: &mut usize,
-) {
-	output.fill(0.0);
-	if state.stopped.load(Ordering::Acquire) {
-		return;
-	}
-
-	let gain = state.gain();
-	let mut output_offset = 0;
-	while output_offset < output.len() {
-		if *cursor == current.len() {
-			match rx.try_recv() {
-				Ok(next) => {
-					*current = next;
-					*cursor = 0;
-					*empty_callbacks = 0;
-				},
-				Err(TryRecvError::Empty) => {
-					*empty_callbacks = 0;
-					break;
-				},
-				Err(TryRecvError::Disconnected) => {
-					*empty_callbacks += 1;
-					if *empty_callbacks >= PLAYBACK_DRAIN_CALLBACKS {
-						state.mark_drained();
-					}
-					break;
-				},
-			}
-		}
-
-		let count = (current.len() - *cursor).min(output.len() - output_offset);
-		let source = &current[*cursor..*cursor + count];
-		let destination = &mut output[output_offset..output_offset + count];
-		if gain == 1.0 {
-			destination.copy_from_slice(source);
-		} else {
-			for (destination, source) in destination.iter_mut().zip(source) {
-				*destination = *source * gain;
-			}
-		}
-		*cursor += count;
-		output_offset += count;
-	}
-}
-
-fn start_capture_device<C>(sample_rate: u32, mut on_audio: C) -> NativeResult<Device<f32>>
-where
-	C: FnMut(&[f32]) + Send + 'static,
-{
-	let sample_rate = audio_sample_rate(sample_rate)?;
-	let mut builder = DeviceBuilder::capture().f32();
-	builder
-		.sample_rate(sample_rate)
-		.capture_channels(AUDIO_CHANNELS)
-		.period_size_millis(CAPTURE_PERIOD_MS)
-		.performance_profile(PerformanceProfile::LowLatency)
-		.backends(AUDIO_BACKENDS);
-	let mut device = builder
-		.with_callback(move |_device, samples| {
-			if !samples.is_empty() {
-				on_audio(samples);
-			}
-		})
-		.map_err(|error| format!("Failed to open the default microphone: {error}"))?;
-	device
-		.device_start()
-		.map_err(|error| format!("Failed to start microphone capture: {error}"))?;
-	Ok(device)
-}
 
 /// Default-microphone capture converted to mono `f32` at the requested sample
 /// rate.
 #[napi]
 pub struct AudioCapture {
-	device: Mutex<Option<Device<f32>>>,
+	stream: Mutex<Option<CaptureStream>>,
 }
 
 #[napi]
@@ -316,32 +32,22 @@ impl AudioCapture {
 		#[napi(ts_arg_type = "(error: Error | null, samples: Float32Array) => void")]
 		on_audio: CaptureCallback,
 	) -> Result<Self> {
-		let device = start_capture_device(sample_rate, move |samples| {
+		let stream = CaptureStream::start(sample_rate, move |samples| {
 			on_audio
 				.call(Ok(Float32Array::new(samples.to_vec())), ThreadsafeFunctionCallMode::NonBlocking);
 		})
 		.map_err(napi::Error::from_reason)?;
-		Ok(Self { device: Mutex::new(Some(device)) })
+		Ok(Self { stream: Mutex::new(Some(stream)) })
 	}
 
 	/// Stop capture immediately and release the microphone.
 	#[napi]
 	pub fn stop(&self) -> Result<()> {
-		let device = self.device.lock().take();
+		let stream = self.stream.lock().take();
-		let Some(mut device) = device else {
+		let Some(mut stream) = stream else {
 			return Ok(());
 		};
-		device.device_stop().map_err(|error| {
+		stream.stop().map_err(napi::Error::from_reason)
-			napi::Error::from_reason(format!("Failed to stop microphone capture: {error}"))
-		})
-	}
-}
-
-impl Drop for AudioCapture {
-	fn drop(&mut self) {
-		if let Some(mut device) = self.device.get_mut().take() {
-			let _ = device.device_stop();
-		}
 	}
 }
 
@@ -416,77 +122,3 @@ impl AudioPlayback {
 		Ok(())
 	}
 }
-
-impl Drop for AudioPlayback {
-	fn drop(&mut self) {
-		if let Some(mut stream) = self.stream.get_mut().take() {
-			let _ = stream.stop();
-		}
-	}
-}
-
-#[cfg(test)]
-mod tests {
-	use std::{
-		env,
-		mem::forget,
-		sync::atomic::AtomicUsize,
-		thread::sleep,
-		time::{Duration, Instant},
-	};
-
-	use super::*;
-
-	#[test]
-	fn playback_preserves_chunk_order_and_applies_render_gain() {
-		let state = PlaybackState::new();
-		state.set_gain(0.5);
-		let (tx, rx) = flume::unbounded();
-		tx.send(vec![1.0, -1.0]).expect("receiver is live");
-		tx.send(vec![0.5, -0.5]).expect("receiver is live");
-		drop(tx);
-		let mut current = Vec::new();
-		let mut cursor = 0;
-		let mut empty_callbacks = 0;
-		let mut output = [9.0; 5];
-
-		fill_playback(&rx, &mut current, &mut cursor, &mut output, &state, &mut empty_callbacks);
-
-		assert_eq!(output, [0.5, -0.5, 0.25, -0.25, 0.0]);
-		assert!(!state.drained.load(Ordering::Acquire));
-		let mut silence = [1.0; 2];
-		while empty_callbacks < PLAYBACK_DRAIN_CALLBACKS {
-			silence.fill(1.0);
-			fill_playback(&rx, &mut current, &mut cursor, &mut silence, &state, &mut empty_callbacks);
-			assert_eq!(silence, [0.0, 0.0]);
-			assert_eq!(
-				state.drained.load(Ordering::Acquire),
-				empty_callbacks >= PLAYBACK_DRAIN_CALLBACKS
-			);
-		}
-	}
-
-	#[test]
-	fn opt_in_default_capture_receives_frames() {
-		if env::var_os("OMP_NATIVE_AUDIO_CAPTURE_TEST").is_none() {
-			return;
-		}
-
-		let callbacks = Arc::new(AtomicUsize::new(0));
-		let callback_count = Arc::clone(&callbacks);
-		let mut device = start_capture_device(16_000, move |_samples| {
-			callback_count.fetch_add(1, Ordering::Relaxed);
-		})
-		.expect("default capture device starts");
-
-		let deadline = Instant::now() + Duration::from_secs(5);
-		while callbacks.load(Ordering::Relaxed) == 0 && Instant::now() < deadline {
-			sleep(Duration::from_millis(20));
-		}
-		if callbacks.load(Ordering::Relaxed) == 0 {
-			forget(device);
-			panic!("capture device started but delivered no frames within five seconds");
-		}
-		device.device_stop().expect("capture device stops");
-	}
-}

crates/pi-natives/src/live.rs

@@ -1,365 +1,21 @@
-//! Native WebRTC media transport for Codex live conversations.
+//! N-API bindings for the Codex live WebRTC peer.
 //!
 //! The TypeScript host owns authenticated signaling and the sideband protocol;
-//! this module owns the realtime WebRTC peer, Opus media, and speaker playback.
+//! the realtime peer, Opus media, and speaker playback live in
+//! `pi_voice::live`. This class adapts its callbacks to non-blocking
+//! threadsafe functions and its PCM input to `Float32Array`.
 
-use std::{
+use std::sync::Arc;
-	sync::{
-		Arc, Weak,
-		atomic::{AtomicBool, AtomicUsize, Ordering},
-	},
-	time::Duration,
-};
 
-use bytes::Bytes;
 use napi::{
 	bindgen_prelude::{Float32Array, Result},
 	threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode, UnknownReturnValue},
 };
 use napi_derive::napi;
-use opus::{Application, Channels, Decoder, Encoder};
+use pi_voice::live::{DEFAULT_OPEN_TIMEOUT_MS, LiveCallbacks, LivePeerCore};
-use parking_lot::Mutex;
-use tokio::{sync::watch, task::JoinHandle};
-use webrtc::{
-	api::{
-		APIBuilder,
-		interceptor_registry::register_default_interceptors,
-		media_engine::{MIME_TYPE_OPUS, MediaEngine},
-	},
-	data_channel::{RTCDataChannel, data_channel_message::DataChannelMessage},
-	interceptor::registry::Registry,
-	media::Sample,
-	peer_connection::{
-		RTCPeerConnection, configuration::RTCConfiguration,
-		peer_connection_state::RTCPeerConnectionState,
-		sdp::session_description::RTCSessionDescription,
-	},
-	rtp_transceiver::{
-		rtp_codec::{RTCRtpCodecCapability, RTCRtpCodecParameters, RTPCodecType},
-		rtp_sender::RTCRtpSender,
-	},
-	track::{
-		track_local::{TrackLocal, track_local_static_sample::TrackLocalStaticSample},
-		track_remote::TrackRemote,
-	},
-};
-
-use crate::audio::{PlaybackStream, PlaybackWriter};
-
-const DATA_CHANNEL_LABEL: &str = "oai-events";
-const INPUT_SAMPLE_RATE: u32 = 16_000;
-const INPUT_FRAME_SAMPLES: usize = 320;
-const INPUT_FRAME_DURATION: Duration = Duration::from_millis(20);
-const MAX_ENCODED_OPUS_BYTES: usize = 1_275;
-const MAX_QUEUED_INPUT_SAMPLES: usize = 32_000;
-const OUTPUT_SAMPLE_RATE: u32 = 48_000;
-const MAX_DECODED_OPUS_SAMPLES: usize = 5_760;
-const OUTPUT_LEVEL_SAMPLES: usize = 2_400;
-const OUTPUT_FRAME_SAMPLES: usize = 960;
-const DEFAULT_OPEN_TIMEOUT_MS: u32 = 20_000;
-const DISCONNECT_GRACE: Duration = Duration::from_secs(2);
-const CLOSE_TASK_TIMEOUT: Duration = Duration::from_secs(1);
-
-const OPUS_CAPABILITY: RTCRtpCodecCapability = RTCRtpCodecCapability {
-	mime_type:     String::new(),
-	clock_rate:    OUTPUT_SAMPLE_RATE,
-	channels:      2,
-	sdp_fmtp_line: String::new(),
-	rtcp_feedback: Vec::new(),
-};
 
 type StringCallback = ThreadsafeFunction<String, UnknownReturnValue>;
 type LevelCallback = ThreadsafeFunction<f64, UnknownReturnValue>;
-type NativeResult<T> = std::result::Result<T, String>;
-
-#[derive(Clone, Debug)]
-enum PeerSignal {
-	Connecting,
-	Open,
-	Failed(String),
-	Closed,
-}
-
-enum InputCommand {
-	Audio(Vec<f32>),
-	Muted(bool),
-	Close,
-}
-
-struct LiveCallbacks {
-	event:   StringCallback,
-	level:   LevelCallback,
-	failure: StringCallback,
-}
-
-struct LiveResources {
-	peer:         Arc<RTCPeerConnection>,
-	data_channel: Arc<RTCDataChannel>,
-	input_tx:     flume::Sender<InputCommand>,
-	input_task:   JoinHandle<()>,
-	rtcp_task:    JoinHandle<()>,
-	playback:     PlaybackStream,
-}
-
-struct LivePeerCore {
-	callbacks:        LiveCallbacks,
-	resources:        Mutex<Option<LiveResources>>,
-	signal_tx:        watch::Sender<PeerSignal>,
-	started:          AtomicBool,
-	closing:          AtomicBool,
-	muted:            AtomicBool,
-	failure_reported: AtomicBool,
-	queued_samples:   AtomicUsize,
-}
-
-impl LivePeerCore {
-	fn new(callbacks: LiveCallbacks) -> Self {
-		let (signal_tx, _) = watch::channel(PeerSignal::Connecting);
-		Self {
-			callbacks,
-			resources: Mutex::new(None),
-			signal_tx,
-			started: AtomicBool::new(false),
-			closing: AtomicBool::new(false),
-			muted: AtomicBool::new(false),
-			failure_reported: AtomicBool::new(false),
-			queued_samples: AtomicUsize::new(0),
-		}
-	}
-
-	async fn create_offer(self: &Arc<Self>) -> NativeResult<String> {
-		if self.started.swap(true, Ordering::AcqRel) {
-			return Err("Native live WebRTC peer has already started".to_owned());
-		}
-		if self.closing.load(Ordering::Acquire) {
-			return Err("Native live WebRTC peer is closed".to_owned());
-		}
-
-		let playback = PlaybackStream::start(OUTPUT_SAMPLE_RATE)?;
-		let playback_tx = playback.writer()?;
-		let mut media_engine = MediaEngine::default();
-		let capability = opus_capability();
-		media_engine
-			.register_codec(
-				RTCRtpCodecParameters {
-					capability: capability.clone(),
-					payload_type: 111,
-					..Default::default()
-				},
-				RTPCodecType::Audio,
-			)
-			.map_err(|error| format!("Failed to register the live Opus codec: {error}"))?;
-		let registry = register_default_interceptors(Registry::new(), &mut media_engine)
-			.map_err(|error| format!("Failed to configure live WebRTC interceptors: {error}"))?;
-		let api = APIBuilder::new()
-			.with_media_engine(media_engine)
-			.with_interceptor_registry(registry)
-			.build();
-		let peer = Arc::new(
-			api.new_peer_connection(RTCConfiguration::default())
-				.await
-				.map_err(|error| format!("Failed to create the live WebRTC peer: {error}"))?,
-		);
-
-		let track = Arc::new(TrackLocalStaticSample::new(
-			capability,
-			"audio".to_owned(),
-			"omp-live".to_owned(),
-		));
-		let sender = match peer
-			.add_track(Arc::clone(&track) as Arc<dyn TrackLocal + Send + Sync>)
-			.await
-		{
-			Ok(sender) => sender,
-			Err(error) => {
-				let _ = peer.close().await;
-				return Err(format!("Failed to add the live audio track: {error}"));
-			},
-		};
-
-		install_peer_callbacks(&peer, Arc::downgrade(self), playback_tx);
-		let data_channel = match peer.create_data_channel(DATA_CHANNEL_LABEL, None).await {
-			Ok(channel) => channel,
-			Err(error) => {
-				let _ = peer.close().await;
-				return Err(format!("Failed to create the live data channel: {error}"));
-			},
-		};
-		install_data_channel_callbacks(&data_channel, Arc::downgrade(self));
-
-		let offer = match peer.create_offer(None).await {
-			Ok(offer) => offer,
-			Err(error) => {
-				let _ = peer.close().await;
-				return Err(format!("Failed to create the live SDP offer: {error}"));
-			},
-		};
-		if let Err(error) = peer.set_local_description(offer.clone()).await {
-			let _ = peer.close().await;
-			return Err(format!("Failed to install the live SDP offer: {error}"));
-		}
-		let mut resources_slot = self.resources.lock();
-		if self.closing.load(Ordering::Acquire) {
-			drop(resources_slot);
-			let _ = peer.close().await;
-			return Err("Native live WebRTC peer was closed while starting".to_owned());
-		}
-
-		let (input_tx, input_rx) = flume::unbounded();
-		let input_task = tokio::spawn(run_input_audio(track, input_rx, Arc::downgrade(self)));
-		let rtcp_task = tokio::spawn(drain_rtcp(sender));
-		let resources =
-			LiveResources { peer, data_channel, input_tx, input_task, rtcp_task, playback };
-		*resources_slot = Some(resources);
-		Ok(offer.sdp)
-	}
-
-	async fn accept_answer(&self, sdp: String) -> NativeResult<()> {
-		let peer = self
-			.resources
-			.lock()
-			.as_ref()
-			.map(|resources| Arc::clone(&resources.peer))
-			.ok_or_else(|| "Native live WebRTC peer has not started".to_owned())?;
-		let answer = RTCSessionDescription::answer(sdp)
-			.map_err(|error| format!("Codex returned an invalid live SDP answer: {error}"))?;
-		peer
-			.set_remote_description(answer)
-			.await
-			.map_err(|error| format!("Failed to install the live SDP answer: {error}"))
-	}
-
-	async fn wait_for_open(&self, timeout_ms: u32) -> NativeResult<()> {
-		let mut signal_rx = self.signal_tx.subscribe();
-		let wait = async {
-			loop {
-				let signal = signal_rx.borrow().clone();
-				match signal {
-					PeerSignal::Open => return Ok(()),
-					PeerSignal::Failed(message) => return Err(message),
-					PeerSignal::Closed => {
-						return Err("Native live WebRTC peer closed before opening".to_owned());
-					},
-					PeerSignal::Connecting => {},
-				}
-				signal_rx
-					.changed()
-					.await
-					.map_err(|_| "Native live WebRTC peer stopped before opening".to_owned())?;
-			}
-		};
-		tokio::time::timeout(Duration::from_millis(u64::from(timeout_ms)), wait)
-			.await
-			.map_err(|_| "Timed out waiting for the live data channel to open".to_owned())?
-	}
-
-	fn push_audio(&self, samples: &[f32]) -> NativeResult<()> {
-		if samples.is_empty() || self.muted.load(Ordering::Acquire) {
-			return Ok(());
-		}
-		let input_tx = self
-			.resources
-			.lock()
-			.as_ref()
-			.map(|resources| resources.input_tx.clone())
-			.ok_or_else(|| "Native live WebRTC peer has not started".to_owned())?;
-		let sample_count = samples.len().min(MAX_QUEUED_INPUT_SAMPLES);
-		let retained = &samples[samples.len() - sample_count..];
-		let queued = self
-			.queued_samples
-			.fetch_add(sample_count, Ordering::AcqRel);
-		if queued.saturating_add(sample_count) > MAX_QUEUED_INPUT_SAMPLES {
-			self
-				.queued_samples
-				.fetch_sub(sample_count, Ordering::AcqRel);
-			return Ok(());
-		}
-		if input_tx
-			.send(InputCommand::Audio(retained.to_vec()))
-			.is_err()
-		{
-			self
-				.queued_samples
-				.fetch_sub(sample_count, Ordering::AcqRel);
-			return Err("Native live audio input is closed".to_owned());
-		}
-		Ok(())
-	}
-
-	fn set_muted(&self, muted: bool) -> NativeResult<()> {
-		self.muted.store(muted, Ordering::Release);
-		let input_tx = self
-			.resources
-			.lock()
-			.as_ref()
-			.map(|resources| resources.input_tx.clone());
-		if let Some(input_tx) = input_tx {
-			input_tx
-				.send(InputCommand::Muted(muted))
-				.map_err(|_| "Native live audio input is closed".to_owned())?;
-		}
-		Ok(())
-	}
-
-	fn report_event(&self, payload: String) {
-		self
-			.callbacks
-			.event
-			.call(Ok(payload), ThreadsafeFunctionCallMode::NonBlocking);
-	}
-
-	fn report_level(&self, level: f64) {
-		self
-			.callbacks
-			.level
-			.call(Ok(level.clamp(0.0, 1.0)), ThreadsafeFunctionCallMode::NonBlocking);
-	}
-
-	fn mark_open(&self) {
-		if !self.closing.load(Ordering::Acquire) {
-			self.signal_tx.send_replace(PeerSignal::Open);
-		}
-	}
-
-	fn report_failure(&self, message: String) {
-		if self.closing.load(Ordering::Acquire) || self.failure_reported.swap(true, Ordering::AcqRel)
-		{
-			return;
-		}
-		self
-			.signal_tx
-			.send_replace(PeerSignal::Failed(message.clone()));
-		self
-			.callbacks
-			.failure
-			.call(Ok(message), ThreadsafeFunctionCallMode::NonBlocking);
-	}
-
-	async fn close(&self) {
-		if self.closing.swap(true, Ordering::AcqRel) {
-			let mut signal_rx = self.signal_tx.subscribe();
-			while !matches!(*signal_rx.borrow(), PeerSignal::Closed) {
-				if signal_rx.changed().await.is_err() {
-					break;
-				}
-			}
-			return;
-		}
-
-		let resources = self.resources.lock().take();
-		if let Some(mut resources) = resources {
-			let _ = resources.input_tx.send(InputCommand::Close);
-			let _ = resources.peer.close().await;
-			let _ = resources.playback.stop();
-			let _ = tokio::time::timeout(CLOSE_TASK_TIMEOUT, resources.input_task).await;
-			resources.rtcp_task.abort();
-			let _ = resources.rtcp_task.await;
-			drop(resources.data_channel);
-		}
-		self.queued_samples.store(0, Ordering::Release);
-		self.signal_tx.send_replace(PeerSignal::Closed);
-	}
-}
 
 /// WebRTC peer that accepts 16 kHz mono PCM and renders remote Opus audio.
 #[napi]
@@ -381,9 +37,15 @@ impl LiveWebRtcPeer {
 	) -> Self {
 		Self {
 			inner: Arc::new(LivePeerCore::new(LiveCallbacks {
-				event:   on_event,
+				event:   Box::new(move |payload| {
-				level:   on_level,
+					on_event.call(Ok(payload), ThreadsafeFunctionCallMode::NonBlocking);
-				failure: on_failure,
+				}),
+				level:   Box::new(move |level| {
+					on_level.call(Ok(level), ThreadsafeFunctionCallMode::NonBlocking);
+				}),
+				failure: Box::new(move |message| {
+					on_failure.call(Ok(message), ThreadsafeFunctionCallMode::NonBlocking);
+				}),
 			})),
 		}
 	}
@@ -446,7 +108,7 @@ impl LiveWebRtcPeer {
 
 impl Drop for LiveWebRtcPeer {
 	fn drop(&mut self) {
-		if self.inner.closing.load(Ordering::Acquire) {
+		if self.inner.is_closing() {
 			return;
 		}
 		let inner = Arc::clone(&self.inner);
@@ -457,345 +119,3 @@ impl Drop for LiveWebRtcPeer {
 		}
 	}
 }
-
-fn opus_capability() -> RTCRtpCodecCapability {
-	RTCRtpCodecCapability {
-		mime_type:     MIME_TYPE_OPUS.to_owned(),
-		clock_rate:    OPUS_CAPABILITY.clock_rate,
-		channels:      OPUS_CAPABILITY.channels,
-		sdp_fmtp_line: "minptime=10;useinbandfec=1".to_owned(),
-		rtcp_feedback: Vec::new(),
-	}
-}
-
-fn install_peer_callbacks(
-	peer: &Arc<RTCPeerConnection>,
-	core: Weak<LivePeerCore>,
-	playback_tx: PlaybackWriter,
-) {
-	let output_sender = Arc::new(Mutex::new(Some(playback_tx)));
-	let output_sender_for_track = Arc::clone(&output_sender);
-	let core_for_track = core.clone();
-	peer.on_track(Box::new(move |track, _receiver, _transceiver| {
-		let output_sender = output_sender_for_track.lock().take();
-		let core = core_for_track.clone();
-		Box::pin(async move {
-			if track.kind() != RTPCodecType::Audio {
-				return;
-			}
-			let Some(output_sender) = output_sender else {
-				if let Some(core) = core.upgrade() {
-					core.report_failure(
-						"Codex live returned more than one remote audio track".to_owned(),
-					);
-				}
-				return;
-			};
-			tokio::spawn(receive_output_audio(track, output_sender, core));
-		})
-	}));
-
-	let peer_for_state = Arc::downgrade(peer);
-	peer.on_peer_connection_state_change(Box::new(move |state| {
-		let core = core.clone();
-		let peer = peer_for_state.clone();
-		Box::pin(async move {
-			let Some(core) = core.upgrade() else {
-				return;
-			};
-			match state {
-				RTCPeerConnectionState::Failed => {
-					core.report_failure("Live WebRTC peer connection failed".to_owned());
-				},
-				RTCPeerConnectionState::Closed if !core.closing.load(Ordering::Acquire) => {
-					core.report_failure("Live WebRTC peer connection closed unexpectedly".to_owned());
-				},
-				RTCPeerConnectionState::Disconnected => {
-					tokio::time::sleep(DISCONNECT_GRACE).await;
-					if peer.upgrade().is_some_and(|peer| {
-						peer.connection_state() == RTCPeerConnectionState::Disconnected
-					}) {
-						core.report_failure("Live WebRTC peer connection disconnected".to_owned());
-					}
-				},
-				_ => {},
-			}
-		})
-	}));
-}
-
-fn install_data_channel_callbacks(data_channel: &Arc<RTCDataChannel>, core: Weak<LivePeerCore>) {
-	let core_for_open = core.clone();
-	data_channel.on_open(Box::new(move || {
-		Box::pin(async move {
-			if let Some(core) = core_for_open.upgrade() {
-				core.mark_open();
-			}
-		})
-	}));
-
-	let core_for_message = core.clone();
-	data_channel.on_message(Box::new(move |message: DataChannelMessage| {
-		let core = core_for_message.clone();
-		Box::pin(async move {
-			if !message.is_string {
-				return;
-			}
-			if let (Some(core), Ok(payload)) =
-				(core.upgrade(), String::from_utf8(message.data.to_vec()))
-			{
-				core.report_event(payload);
-			}
-		})
-	}));
-
-	let core_for_close = core.clone();
-	data_channel.on_close(Box::new(move || {
-		let core = core_for_close.clone();
-		Box::pin(async move {
-			if let Some(core) = core.upgrade() {
-				core.report_failure("Live data channel closed unexpectedly".to_owned());
-			}
-		})
-	}));
-
-	data_channel.on_error(Box::new(move |error| {
-		let core = core.clone();
-		Box::pin(async move {
-			if let Some(core) = core.upgrade() {
-				core.report_failure(format!("Live data channel failed: {error}"));
-			}
-		})
-	}));
-}
-
-async fn run_input_audio(
-	track: Arc<TrackLocalStaticSample>,
-	input_rx: flume::Receiver<InputCommand>,
-	core: Weak<LivePeerCore>,
-) {
-	let mut encoder = match Encoder::new(INPUT_SAMPLE_RATE, Channels::Mono, Application::Voip) {
-		Ok(encoder) => encoder,
-		Err(error) => {
-			if let Some(core) = core.upgrade() {
-				core.report_failure(format!("Failed to initialize the live Opus encoder: {error}"));
-			}
-			return;
-		},
-	};
-	if let Err(error) = encoder.set_inband_fec(true) {
-		if let Some(core) = core.upgrade() {
-			core.report_failure(format!("Failed to configure the live Opus encoder: {error}"));
-		}
-		return;
-	}
-
-	let mut muted = false;
-	let mut pending = Vec::with_capacity(INPUT_FRAME_SAMPLES * 2);
-	let mut encoded = [0u8; MAX_ENCODED_OPUS_BYTES];
-	let mut ticker = tokio::time::interval(INPUT_FRAME_DURATION);
-	ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Burst);
-	ticker.tick().await;
-	loop {
-		tokio::select! {
-			biased;
-			command = input_rx.recv_async() => {
-				let Ok(command) = command else {
-					break;
-				};
-				match command {
-					InputCommand::Audio(samples) => {
-						if let Some(core) = core.upgrade() {
-							core.queued_samples.fetch_sub(samples.len(), Ordering::AcqRel);
-						}
-						if muted {
-							continue;
-						}
-						if samples.len() >= MAX_QUEUED_INPUT_SAMPLES {
-							pending.clear();
-							pending.extend_from_slice(&samples[samples.len() - MAX_QUEUED_INPUT_SAMPLES..]);
-							continue;
-						}
-						let overflow = pending
-							.len()
-							.saturating_add(samples.len())
-							.saturating_sub(MAX_QUEUED_INPUT_SAMPLES);
-						if overflow > 0 {
-							pending.drain(..overflow);
-						}
-						pending.extend_from_slice(&samples);
-					},
-					InputCommand::Muted(next_muted) => {
-						muted = next_muted;
-						pending.clear();
-					},
-					InputCommand::Close => break,
-				}
-			},
-			_ = ticker.tick() => {
-				let mut frame = [0.0f32; INPUT_FRAME_SAMPLES];
-				if !muted {
-					let consumed = pending.len().min(INPUT_FRAME_SAMPLES);
-					frame[..consumed].copy_from_slice(&pending[..consumed]);
-					if consumed > 0 {
-						pending.copy_within(consumed.., 0);
-						pending.truncate(pending.len() - consumed);
-					}
-				}
-				let encoded_len = match encoder.encode_float(&frame, &mut encoded) {
-					Ok(encoded_len) => encoded_len,
-					Err(error) => {
-						if let Some(core) = core.upgrade() {
-							core.report_failure(format!("Failed to encode live microphone audio: {error}"));
-						}
-						return;
-					},
-				};
-				let sample = Sample {
-					data: Bytes::copy_from_slice(&encoded[..encoded_len]),
-					duration: INPUT_FRAME_DURATION,
-					..Default::default()
-				};
-				if let Err(error) = track.write_sample(&sample).await {
-					if let Some(core) = core.upgrade() {
-						core.report_failure(format!("Failed to send live microphone audio: {error}"));
-					}
-					return;
-				}
-			},
-		}
-	}
-}
-
-async fn drain_rtcp(sender: Arc<RTCRtpSender>) {
-	while sender.read_rtcp().await.is_ok() {}
-}
-
-async fn receive_output_audio(
-	track: Arc<TrackRemote>,
-	playback_tx: PlaybackWriter,
-	core: Weak<LivePeerCore>,
-) {
-	if !track
-		.codec()
-		.capability
-		.mime_type
-		.eq_ignore_ascii_case(MIME_TYPE_OPUS)
-	{
-		if let Some(core) = core.upgrade() {
-			core.report_failure(format!(
-				"Codex live negotiated unsupported audio codec {}",
-				track.codec().capability.mime_type
-			));
-		}
-		return;
-	}
-	let mut decoder = match Decoder::new(OUTPUT_SAMPLE_RATE, Channels::Mono) {
-		Ok(decoder) => decoder,
-		Err(error) => {
-			if let Some(core) = core.upgrade() {
-				core.report_failure(format!("Failed to initialize the live Opus decoder: {error}"));
-			}
-			return;
-		},
-	};
-	let mut decoded = vec![0.0f32; MAX_DECODED_OPUS_SAMPLES].into_boxed_slice();
-	let mut expected_sequence: Option<u16> = None;
-	let mut level = OutputLevel::default();
-
-	loop {
-		let packet = match track.read_rtp().await {
-			Ok((packet, _attributes)) => packet,
-			Err(error) => {
-				if let Some(core) = core.upgrade()
-					&& !core.closing.load(Ordering::Acquire)
-				{
-					core.report_failure(format!("Live remote audio track failed: {error}"));
-				}
-				return;
-			},
-		};
-		let sequence = packet.header.sequence_number;
-		if let Some(expected) = expected_sequence {
-			let gap = sequence.wrapping_sub(expected);
-			if gap >= u16::MAX / 2 {
-				continue;
-			}
-			if gap > 0 {
-				for _ in 1..gap.min(5) {
-					if let Ok(samples) =
-						decoder.decode_float(&[], &mut decoded[..OUTPUT_FRAME_SAMPLES], false)
-					{
-						if !write_output(&playback_tx, &decoded[..samples], &core) {
-							return;
-						}
-						level.observe(&decoded[..samples], &core);
-					}
-				}
-				if let Ok(samples) = decoder.decode_float(&packet.payload, &mut decoded, true) {
-					if !write_output(&playback_tx, &decoded[..samples], &core) {
-						return;
-					}
-					level.observe(&decoded[..samples], &core);
-				}
-			}
-		}
-		expected_sequence = Some(sequence.wrapping_add(1));
-		match decoder.decode_float(&packet.payload, &mut decoded, false) {
-			Ok(samples) => {
-				if !write_output(&playback_tx, &decoded[..samples], &core) {
-					return;
-				}
-				level.observe(&decoded[..samples], &core);
-			},
-			Err(error) => {
-				if let Some(core) = core.upgrade() {
-					core.report_failure(format!("Failed to decode live speaker audio: {error}"));
-				}
-				return;
-			},
-		}
-	}
-}
-
-fn write_output(playback_tx: &PlaybackWriter, samples: &[f32], core: &Weak<LivePeerCore>) -> bool {
-	match playback_tx.write(samples) {
-		Ok(()) => true,
-		Err(error) => {
-			if let Some(core) = core.upgrade()
-				&& !core.closing.load(Ordering::Acquire)
-			{
-				core.report_failure(format!("Live speaker playback failed: {error}"));
-			}
-			false
-		},
-	}
-}
-
-#[derive(Default)]
-struct OutputLevel {
-	sum_squares: f64,
-	samples:     usize,
-}
-
-impl OutputLevel {
-	fn observe(&mut self, decoded: &[f32], core: &Weak<LivePeerCore>) {
-		let mut offset = 0;
-		while offset < decoded.len() {
-			let take = (OUTPUT_LEVEL_SAMPLES - self.samples).min(decoded.len() - offset);
-			for &sample in &decoded[offset..offset + take] {
-				let sample = f64::from(sample);
-				self.sum_squares = sample.mul_add(sample, self.sum_squares);
-			}
-			self.samples += take;
-			offset += take;
-			if self.samples == OUTPUT_LEVEL_SAMPLES {
-				if let Some(core) = core.upgrade() {
-					core.report_level((self.sum_squares / self.samples as f64).sqrt());
-				}
-				self.sum_squares = 0.0;
-				self.samples = 0;
-			}
-		}
-	}
-}

crates/pi-voice/BUILD.bazel

@@ -0,0 +1,29 @@
+load("@crates//:defs.bzl", "aliases", "all_crate_deps")
+load("@rules_rust//rust:defs.bzl", "rust_library", "rust_test")
+
+package(default_visibility = ["//visibility:public"])
+
+exports_files(["Cargo.toml"])
+
+# napi-free voice engine (miniaudio capture/playback + WebRTC/Opus live peer).
+# Keeping the webrtc/opus/maudio graph behind this rlib means the pi_natives
+# leaf — which recompiles on every release (version-sentinel edit in lib.rs) —
+# no longer pays for the voice stack's monomorphization.
+rust_library(
+    name = "pi-voice",
+    srcs = glob(["src/**/*.rs"]),
+    aliases = aliases(),
+    crate_name = "pi_voice",
+    edition = "2024",
+    proc_macro_deps = all_crate_deps(proc_macro = True),
+    version = "17.1.5",
+    deps = all_crate_deps(normal = True),
+)
+
+rust_test(
+    name = "pi-voice_test",
+    crate = ":pi-voice",
+    edition = "2024",
+    proc_macro_deps = all_crate_deps(proc_macro_dev = True),
+    deps = all_crate_deps(normal_dev = True),
+)

crates/pi-voice/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "pi-voice"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+authors.workspace = true
+repository.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+# Forces the static libopus build for the `opus` codec crate.
+audiopus_sys.workspace = true
+bytes.workspace = true
+flume.workspace = true
+maudio.workspace = true
+opus.workspace = true
+parking_lot.workspace = true
+tokio.workspace = true
+webrtc.workspace = true
+
+[target.'cfg(target_os = "macos")'.dependencies]
+# Regenerate miniaudio FFI bindings from the compiled headers on macOS only.
+# The maudio-sys pregenerated `unix.rs` bindings are Linux-shaped: their
+# `ma_device`/`ma_context` backend-state unions omit the `#ifdef
+# MA_SUPPORT_COREAUDIO` members that the macOS build of miniaudio.c actually
+# contains, so shipping them on darwin is an ABI mismatch that silently breaks
+# CoreAudio capture (device init/start succeed but no render callbacks fire).
+# darwin builds with host Xcode (libclang available), so bindgen is fine here;
+# the hermetic zig/musl/windows toolchains keep the bindgen-free pregen path,
+# whose bindings are correct for those platforms.
+maudio = { workspace = true, features = ["generate-bindings"] }

crates/pi-voice/src/audio.rs

@@ -0,0 +1,400 @@
+//! Cross-platform microphone capture and streaming speaker playback.
+//!
+//! miniaudio owns platform device discovery, format conversion, channel mixing,
+//! and resampling. The engine exposes one stable mono `f32` contract: the
+//! N-API classes in pi-natives adapt it to TypeScript, and [`crate::live`]
+//! shares [`PlaybackStream`] for remote-audio rendering.
+
+use std::sync::{
+	Arc,
+	atomic::{AtomicBool, AtomicU32, Ordering},
+};
+
+use flume::TryRecvError;
+use maudio::{
+	audio::{performance::PerformanceProfile, sample_rate::SampleRate},
+	backend::Backend,
+	device::{
+		Device,
+		device_builder::{DeviceBuilder, DeviceBuilderOps},
+	},
+};
+use tokio::sync::Notify;
+
+use crate::VoiceResult;
+
+const AUDIO_CHANNELS: u32 = 1;
+// PulseAudio TCP playback stutters with a 20 ms target buffer; 50 ms absorbs
+// transport jitter while preserving interactive latency.
+#[cfg(target_os = "linux")]
+const PLAYBACK_PERIOD_MS: u32 = 50;
+#[cfg(not(target_os = "linux"))]
+const PLAYBACK_PERIOD_MS: u32 = 20;
+// miniaudio's PulseAudio backend reserves three periods. Android's OpenSL ES
+// source emits 125 ms fragments, so Linux capture needs at least 150 ms queued.
+#[cfg(target_os = "linux")]
+const CAPTURE_PERIOD_MS: u32 = 50;
+#[cfg(not(target_os = "linux"))]
+const CAPTURE_PERIOD_MS: u32 = 20;
+// PulseAudio can retain its default three periods after the producer closes.
+// Wait for all of them before stopping the device so the tail reaches the sink.
+#[cfg(target_os = "linux")]
+const PLAYBACK_DRAIN_CALLBACKS: usize = 3;
+#[cfg(not(target_os = "linux"))]
+const PLAYBACK_DRAIN_CALLBACKS: usize = 2;
+
+#[cfg(target_os = "macos")]
+const AUDIO_BACKENDS: &[Backend] = &[Backend::CoreAudio];
+#[cfg(target_os = "windows")]
+const AUDIO_BACKENDS: &[Backend] = &[Backend::Wasapi];
+#[cfg(target_os = "linux")]
+const AUDIO_BACKENDS: &[Backend] = &[Backend::PulseAudio, Backend::Alsa, Backend::Jack];
+#[cfg(not(any(target_os = "macos", target_os = "windows", target_os = "linux")))]
+const AUDIO_BACKENDS: &[Backend] = &[Backend::Sndio, Backend::Audio4, Backend::Oss];
+
+/// Shared render-time state for one playback device: gain, drain, stop.
+///
+/// Held as an `Arc` by both the stream and its N-API wrapper so
+/// [`PlaybackState::wait_for_drain`] can outlive the stream lock.
+pub struct PlaybackState {
+	gain_bits: AtomicU32,
+	drained:   AtomicBool,
+	stopped:   AtomicBool,
+	notify:    Notify,
+}
+
+impl PlaybackState {
+	fn new() -> Self {
+		Self {
+			gain_bits: AtomicU32::new(1.0f32.to_bits()),
+			drained:   AtomicBool::new(false),
+			stopped:   AtomicBool::new(false),
+			notify:    Notify::new(),
+		}
+	}
+
+	fn gain(&self) -> f32 {
+		f32::from_bits(self.gain_bits.load(Ordering::Acquire))
+	}
+
+	fn set_gain(&self, gain: f32) {
+		self.gain_bits.store(gain.to_bits(), Ordering::Release);
+	}
+
+	fn mark_drained(&self) {
+		if !self.drained.swap(true, Ordering::AcqRel) {
+			self.notify.notify_waiters();
+		}
+	}
+
+	fn mark_stopped(&self) {
+		self.stopped.store(true, Ordering::Release);
+		self.notify.notify_waiters();
+	}
+
+	/// Resolve once every queued sample reached the speaker (or the stream
+	/// stopped). Used by the N-API `AudioPlayback.end()` graceful-close path.
+	pub async fn wait_for_drain(&self) {
+		loop {
+			let notified = self.notify.notified();
+			if self.drained.load(Ordering::Acquire) || self.stopped.load(Ordering::Acquire) {
+				return;
+			}
+			notified.await;
+		}
+	}
+}
+
+/// Producer endpoint for one native playback device. Cloned into the WebRTC
+/// remote-audio decoder so it can feed the same speaker stream.
+#[derive(Clone)]
+pub struct PlaybackWriter {
+	tx:    flume::Sender<Vec<f32>>,
+	state: Arc<PlaybackState>,
+}
+
+impl PlaybackWriter {
+	/// Queue mono floating-point samples without blocking the caller.
+	pub fn write(&self, samples: &[f32]) -> VoiceResult<()> {
+		if samples.is_empty() {
+			return Ok(());
+		}
+		if self.state.stopped.load(Ordering::Acquire) || self.state.drained.load(Ordering::Acquire) {
+			return Err("Native audio playback is closed".to_owned());
+		}
+		self
+			.tx
+			.send(samples.to_vec())
+			.map_err(|_| "Native audio playback is closed".to_owned())
+	}
+}
+
+/// Running mono playback stream shared by N-API playback and native WebRTC.
+pub struct PlaybackStream {
+	device: Option<Device<f32>>,
+	writer: Option<PlaybackWriter>,
+	state:  Arc<PlaybackState>,
+}
+
+impl PlaybackStream {
+	/// Open and start the default speaker at the requested logical sample rate.
+	pub fn start(sample_rate: u32) -> VoiceResult<Self> {
+		let sample_rate = audio_sample_rate(sample_rate)?;
+		let state = Arc::new(PlaybackState::new());
+		let (tx, rx) = flume::unbounded::<Vec<f32>>();
+		let callback_state = Arc::clone(&state);
+		let mut current = Vec::new();
+		let mut cursor = 0;
+		let mut empty_callbacks = 0;
+		let mut builder = DeviceBuilder::playback().f32();
+		builder
+			.sample_rate(sample_rate)
+			.playback_channels(AUDIO_CHANNELS)
+			.period_size_millis(PLAYBACK_PERIOD_MS)
+			.performance_profile(PerformanceProfile::LowLatency)
+			.backends(AUDIO_BACKENDS);
+		let mut device = builder
+			.with_callback(move |_device, output| {
+				fill_playback(
+					&rx,
+					&mut current,
+					&mut cursor,
+					output,
+					&callback_state,
+					&mut empty_callbacks,
+				);
+			})
+			.map_err(|error| format!("Failed to open the default speaker: {error}"))?;
+		device
+			.device_start()
+			.map_err(|error| format!("Failed to start speaker playback: {error}"))?;
+
+		Ok(Self {
+			device: Some(device),
+			writer: Some(PlaybackWriter { tx, state: Arc::clone(&state) }),
+			state,
+		})
+	}
+
+	/// Clone the producer endpoint used by the remote-audio decoder.
+	pub fn writer(&self) -> VoiceResult<PlaybackWriter> {
+		self
+			.writer
+			.clone()
+			.ok_or_else(|| "Native audio playback is closed".to_owned())
+	}
+
+	/// Shared render-time state, cloned out so callers can await drain after
+	/// releasing the stream lock.
+	pub fn state(&self) -> Arc<PlaybackState> {
+		Arc::clone(&self.state)
+	}
+
+	/// Close the producer side so the render callback can detect drain.
+	pub fn finish_input(&mut self) {
+		self.writer.take();
+	}
+
+	/// Scale audio at render time so gain changes affect already queued
+	/// samples. Rejects non-finite gains; negative gains clamp to silence.
+	pub fn set_gain(&self, gain: f32) -> VoiceResult<()> {
+		if !gain.is_finite() {
+			return Err("Audio playback gain must be finite".to_owned());
+		}
+		self.state.set_gain(gain.max(0.0));
+		Ok(())
+	}
+
+	/// Stop playback immediately and release the default speaker.
+	pub fn stop(&mut self) -> VoiceResult<()> {
+		self.writer.take();
+		self.state.mark_stopped();
+		let Some(mut device) = self.device.take() else {
+			return Ok(());
+		};
+		device
+			.device_stop()
+			.map_err(|error| format!("Failed to stop speaker playback: {error}"))
+	}
+}
+
+impl Drop for PlaybackStream {
+	fn drop(&mut self) {
+		let _ = self.stop();
+	}
+}
+
+fn audio_sample_rate(sample_rate: u32) -> VoiceResult<SampleRate> {
+	SampleRate::try_from(sample_rate)
+		.map_err(|error| format!("Unsupported audio sample rate {sample_rate}: {error}"))
+}
+
+fn fill_playback(
+	rx: &flume::Receiver<Vec<f32>>,
+	current: &mut Vec<f32>,
+	cursor: &mut usize,
+	output: &mut [f32],
+	state: &PlaybackState,
+	empty_callbacks: &mut usize,
+) {
+	output.fill(0.0);
+	if state.stopped.load(Ordering::Acquire) {
+		return;
+	}
+
+	let gain = state.gain();
+	let mut output_offset = 0;
+	while output_offset < output.len() {
+		if *cursor == current.len() {
+			match rx.try_recv() {
+				Ok(next) => {
+					*current = next;
+					*cursor = 0;
+					*empty_callbacks = 0;
+				},
+				Err(TryRecvError::Empty) => {
+					*empty_callbacks = 0;
+					break;
+				},
+				Err(TryRecvError::Disconnected) => {
+					*empty_callbacks += 1;
+					if *empty_callbacks >= PLAYBACK_DRAIN_CALLBACKS {
+						state.mark_drained();
+					}
+					break;
+				},
+			}
+		}
+
+		let count = (current.len() - *cursor).min(output.len() - output_offset);
+		let source = &current[*cursor..*cursor + count];
+		let destination = &mut output[output_offset..output_offset + count];
+		if gain == 1.0 {
+			destination.copy_from_slice(source);
+		} else {
+			for (destination, source) in destination.iter_mut().zip(source) {
+				*destination = *source * gain;
+			}
+		}
+		*cursor += count;
+		output_offset += count;
+	}
+}
+
+/// Running default-microphone capture delivering low-latency mono `f32`
+/// chunks to its callback. Wraps the miniaudio device so N-API callers never
+/// see maudio types.
+pub struct CaptureStream {
+	device: Option<Device<f32>>,
+}
+
+impl CaptureStream {
+	/// Open the default microphone at the requested sample rate. `on_audio`
+	/// runs on the realtime audio thread — it must not block.
+	pub fn start<C>(sample_rate: u32, mut on_audio: C) -> VoiceResult<Self>
+	where
+		C: FnMut(&[f32]) + Send + 'static,
+	{
+		let sample_rate = audio_sample_rate(sample_rate)?;
+		let mut builder = DeviceBuilder::capture().f32();
+		builder
+			.sample_rate(sample_rate)
+			.capture_channels(AUDIO_CHANNELS)
+			.period_size_millis(CAPTURE_PERIOD_MS)
+			.performance_profile(PerformanceProfile::LowLatency)
+			.backends(AUDIO_BACKENDS);
+		let mut device = builder
+			.with_callback(move |_device, samples| {
+				if !samples.is_empty() {
+					on_audio(samples);
+				}
+			})
+			.map_err(|error| format!("Failed to open the default microphone: {error}"))?;
+		device
+			.device_start()
+			.map_err(|error| format!("Failed to start microphone capture: {error}"))?;
+		Ok(Self { device: Some(device) })
+	}
+
+	/// Stop capture immediately and release the microphone.
+	pub fn stop(&mut self) -> VoiceResult<()> {
+		let Some(mut device) = self.device.take() else {
+			return Ok(());
+		};
+		device
+			.device_stop()
+			.map_err(|error| format!("Failed to stop microphone capture: {error}"))
+	}
+}
+
+impl Drop for CaptureStream {
+	fn drop(&mut self) {
+		let _ = self.stop();
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use std::{
+		env,
+		mem::forget,
+		sync::atomic::AtomicUsize,
+		thread::sleep,
+		time::{Duration, Instant},
+	};
+
+	use super::*;
+
+	#[test]
+	fn playback_preserves_chunk_order_and_applies_render_gain() {
+		let state = PlaybackState::new();
+		state.set_gain(0.5);
+		let (tx, rx) = flume::unbounded();
+		tx.send(vec![1.0, -1.0]).expect("receiver is live");
+		tx.send(vec![0.5, -0.5]).expect("receiver is live");
+		drop(tx);
+		let mut current = Vec::new();
+		let mut cursor = 0;
+		let mut empty_callbacks = 0;
+		let mut output = [9.0; 5];
+
+		fill_playback(&rx, &mut current, &mut cursor, &mut output, &state, &mut empty_callbacks);
+
+		assert_eq!(output, [0.5, -0.5, 0.25, -0.25, 0.0]);
+		assert!(!state.drained.load(Ordering::Acquire));
+		let mut silence = [1.0; 2];
+		while empty_callbacks < PLAYBACK_DRAIN_CALLBACKS {
+			silence.fill(1.0);
+			fill_playback(&rx, &mut current, &mut cursor, &mut silence, &state, &mut empty_callbacks);
+			assert_eq!(silence, [0.0, 0.0]);
+			assert_eq!(
+				state.drained.load(Ordering::Acquire),
+				empty_callbacks >= PLAYBACK_DRAIN_CALLBACKS
+			);
+		}
+	}
+
+	#[test]
+	fn opt_in_default_capture_receives_frames() {
+		if env::var_os("OMP_NATIVE_AUDIO_CAPTURE_TEST").is_none() {
+			return;
+		}
+
+		let callbacks = Arc::new(AtomicUsize::new(0));
+		let callback_count = Arc::clone(&callbacks);
+		let mut stream = CaptureStream::start(16_000, move |_samples| {
+			callback_count.fetch_add(1, Ordering::Relaxed);
+		})
+		.expect("default capture device starts");
+
+		let deadline = Instant::now() + Duration::from_secs(5);
+		while callbacks.load(Ordering::Relaxed) == 0 && Instant::now() < deadline {
+			sleep(Duration::from_millis(20));
+		}
+		if callbacks.load(Ordering::Relaxed) == 0 {
+			forget(stream);
+			panic!("capture device started but delivered no frames within five seconds");
+		}
+		stream.stop().expect("capture device stops");
+	}
+}

crates/pi-voice/src/lib.rs

@@ -0,0 +1,26 @@
+//! Native voice engine: microphone capture, speaker playback, and the WebRTC
+//! live-conversation peer.
+//!
+//! napi-free by design. `pi-natives` wraps these types in thin `#[napi]`
+//! adapters, so the webrtc/opus/miniaudio dependency graph compiles once into
+//! this rlib and rebuilds only when voice code changes — never when the
+//! N-API surface elsewhere in the addon does.
+//!
+//! # Architecture
+//! ```text
+//! JS (packages/natives) -> #[napi] adapters (pi-natives audio.rs / live.rs)
+//!   -> pi_voice::audio  (miniaudio capture/playback engine)
+//!   -> pi_voice::live   (WebRTC peer + Opus media, feeds audio playback)
+//! ```
+//!
+//! Async entry points ([`live::LivePeerCore::create_offer`] and friends) run
+//! on the caller's ambient tokio runtime — napi-rs's `tokio_rt` inside the
+//! addon. Callback fields ([`live::LiveCallbacks`]) are invoked from tokio
+//! worker threads and must not block.
+
+pub mod audio;
+pub mod live;
+
+/// Engine-level result: a human-readable failure message the N-API layer maps
+/// to a JS error verbatim.
+pub type VoiceResult<T> = std::result::Result<T, String>;

crates/pi-voice/src/live.rs

@@ -0,0 +1,722 @@
+//! Native WebRTC media transport for Codex live conversations.
+//!
+//! The TypeScript host owns authenticated signaling and the sideband protocol;
+//! this module owns the realtime WebRTC peer, Opus media, and speaker
+//! playback. The N-API class in pi-natives is a thin adapter over
+//! [`LivePeerCore`].
+
+use std::{
+	sync::{
+		Arc, Weak,
+		atomic::{AtomicBool, AtomicUsize, Ordering},
+	},
+	time::Duration,
+};
+
+use bytes::Bytes;
+use opus::{Application, Channels, Decoder, Encoder};
+use parking_lot::Mutex;
+use tokio::{sync::watch, task::JoinHandle};
+use webrtc::{
+	api::{
+		APIBuilder,
+		interceptor_registry::register_default_interceptors,
+		media_engine::{MIME_TYPE_OPUS, MediaEngine},
+	},
+	data_channel::{RTCDataChannel, data_channel_message::DataChannelMessage},
+	interceptor::registry::Registry,
+	media::Sample,
+	peer_connection::{
+		RTCPeerConnection, configuration::RTCConfiguration,
+		peer_connection_state::RTCPeerConnectionState,
+		sdp::session_description::RTCSessionDescription,
+	},
+	rtp_transceiver::{
+		rtp_codec::{RTCRtpCodecCapability, RTCRtpCodecParameters, RTPCodecType},
+		rtp_sender::RTCRtpSender,
+	},
+	track::{
+		track_local::{TrackLocal, track_local_static_sample::TrackLocalStaticSample},
+		track_remote::TrackRemote,
+	},
+};
+
+use crate::{
+	VoiceResult,
+	audio::{PlaybackStream, PlaybackWriter},
+};
+
+const DATA_CHANNEL_LABEL: &str = "oai-events";
+const INPUT_SAMPLE_RATE: u32 = 16_000;
+const INPUT_FRAME_SAMPLES: usize = 320;
+const INPUT_FRAME_DURATION: Duration = Duration::from_millis(20);
+const MAX_ENCODED_OPUS_BYTES: usize = 1_275;
+const MAX_QUEUED_INPUT_SAMPLES: usize = 32_000;
+const OUTPUT_SAMPLE_RATE: u32 = 48_000;
+const MAX_DECODED_OPUS_SAMPLES: usize = 5_760;
+const OUTPUT_LEVEL_SAMPLES: usize = 2_400;
+const OUTPUT_FRAME_SAMPLES: usize = 960;
+/// Default `wait_for_open` timeout, exposed so the N-API adapter can apply it
+/// when TypeScript passes no override.
+pub const DEFAULT_OPEN_TIMEOUT_MS: u32 = 20_000;
+const DISCONNECT_GRACE: Duration = Duration::from_secs(2);
+const CLOSE_TASK_TIMEOUT: Duration = Duration::from_secs(1);
+
+const OPUS_CAPABILITY: RTCRtpCodecCapability = RTCRtpCodecCapability {
+	mime_type:     String::new(),
+	clock_rate:    OUTPUT_SAMPLE_RATE,
+	channels:      2,
+	sdp_fmtp_line: String::new(),
+	rtcp_feedback: Vec::new(),
+};
+
+#[derive(Clone, Debug)]
+enum PeerSignal {
+	Connecting,
+	Open,
+	Failed(String),
+	Closed,
+}
+
+enum InputCommand {
+	Audio(Vec<f32>),
+	Muted(bool),
+	Close,
+}
+
+/// Host callbacks for peer lifecycle and media events. Every callback is
+/// invoked from tokio worker threads and must not block (the N-API adapter
+/// forwards through non-blocking threadsafe functions).
+pub struct LiveCallbacks {
+	/// One `oai-events` data-channel text payload.
+	pub event:   Box<dyn Fn(String) + Send + Sync>,
+	/// RMS output level in `[0, 1]`, one report per level window.
+	pub level:   Box<dyn Fn(f64) + Send + Sync>,
+	/// Terminal transport failure; reported at most once per peer.
+	pub failure: Box<dyn Fn(String) + Send + Sync>,
+}
+
+struct LiveResources {
+	peer:         Arc<RTCPeerConnection>,
+	data_channel: Arc<RTCDataChannel>,
+	input_tx:     flume::Sender<InputCommand>,
+	input_task:   JoinHandle<()>,
+	rtcp_task:    JoinHandle<()>,
+	playback:     PlaybackStream,
+}
+
+/// WebRTC live-conversation peer: accepts 16 kHz mono PCM input and renders
+/// remote Opus audio to the default speaker. Owned as an `Arc` by the N-API
+/// `LiveWebRtcPeer` wrapper.
+pub struct LivePeerCore {
+	callbacks:        LiveCallbacks,
+	resources:        Mutex<Option<LiveResources>>,
+	signal_tx:        watch::Sender<PeerSignal>,
+	started:          AtomicBool,
+	closing:          AtomicBool,
+	muted:            AtomicBool,
+	failure_reported: AtomicBool,
+	queued_samples:   AtomicUsize,
+}
+
+impl LivePeerCore {
+	/// Create an idle peer with its host callbacks registered.
+	pub fn new(callbacks: LiveCallbacks) -> Self {
+		let (signal_tx, _) = watch::channel(PeerSignal::Connecting);
+		Self {
+			callbacks,
+			resources: Mutex::new(None),
+			signal_tx,
+			started: AtomicBool::new(false),
+			closing: AtomicBool::new(false),
+			muted: AtomicBool::new(false),
+			failure_reported: AtomicBool::new(false),
+			queued_samples: AtomicUsize::new(0),
+		}
+	}
+
+	/// Start the native media peer and return its SDP offer.
+	///
+	/// Fails when called twice, after close, or when the speaker, codec,
+	/// peer, track, or data channel cannot be set up.
+	pub async fn create_offer(self: &Arc<Self>) -> VoiceResult<String> {
+		if self.started.swap(true, Ordering::AcqRel) {
+			return Err("Native live WebRTC peer has already started".to_owned());
+		}
+		if self.closing.load(Ordering::Acquire) {
+			return Err("Native live WebRTC peer is closed".to_owned());
+		}
+
+		let playback = PlaybackStream::start(OUTPUT_SAMPLE_RATE)?;
+		let playback_tx = playback.writer()?;
+		let mut media_engine = MediaEngine::default();
+		let capability = opus_capability();
+		media_engine
+			.register_codec(
+				RTCRtpCodecParameters {
+					capability: capability.clone(),
+					payload_type: 111,
+					..Default::default()
+				},
+				RTPCodecType::Audio,
+			)
+			.map_err(|error| format!("Failed to register the live Opus codec: {error}"))?;
+		let registry = register_default_interceptors(Registry::new(), &mut media_engine)
+			.map_err(|error| format!("Failed to configure live WebRTC interceptors: {error}"))?;
+		let api = APIBuilder::new()
+			.with_media_engine(media_engine)
+			.with_interceptor_registry(registry)
+			.build();
+		let peer = Arc::new(
+			api.new_peer_connection(RTCConfiguration::default())
+				.await
+				.map_err(|error| format!("Failed to create the live WebRTC peer: {error}"))?,
+		);
+
+		let track = Arc::new(TrackLocalStaticSample::new(
+			capability,
+			"audio".to_owned(),
+			"omp-live".to_owned(),
+		));
+		let sender = match peer
+			.add_track(Arc::clone(&track) as Arc<dyn TrackLocal + Send + Sync>)
+			.await
+		{
+			Ok(sender) => sender,
+			Err(error) => {
+				let _ = peer.close().await;
+				return Err(format!("Failed to add the live audio track: {error}"));
+			},
+		};
+
+		install_peer_callbacks(&peer, Arc::downgrade(self), playback_tx);
+		let data_channel = match peer.create_data_channel(DATA_CHANNEL_LABEL, None).await {
+			Ok(channel) => channel,
+			Err(error) => {
+				let _ = peer.close().await;
+				return Err(format!("Failed to create the live data channel: {error}"));
+			},
+		};
+		install_data_channel_callbacks(&data_channel, Arc::downgrade(self));
+
+		let offer = match peer.create_offer(None).await {
+			Ok(offer) => offer,
+			Err(error) => {
+				let _ = peer.close().await;
+				return Err(format!("Failed to create the live SDP offer: {error}"));
+			},
+		};
+		if let Err(error) = peer.set_local_description(offer.clone()).await {
+			let _ = peer.close().await;
+			return Err(format!("Failed to install the live SDP offer: {error}"));
+		}
+		let mut resources_slot = self.resources.lock();
+		if self.closing.load(Ordering::Acquire) {
+			drop(resources_slot);
+			let _ = peer.close().await;
+			return Err("Native live WebRTC peer was closed while starting".to_owned());
+		}
+
+		let (input_tx, input_rx) = flume::unbounded();
+		let input_task = tokio::spawn(run_input_audio(track, input_rx, Arc::downgrade(self)));
+		let rtcp_task = tokio::spawn(drain_rtcp(sender));
+		let resources =
+			LiveResources { peer, data_channel, input_tx, input_task, rtcp_task, playback };
+		*resources_slot = Some(resources);
+		Ok(offer.sdp)
+	}
+
+	/// Apply the remote SDP answer returned by Codex signaling.
+	pub async fn accept_answer(&self, sdp: String) -> VoiceResult<()> {
+		let peer = self
+			.resources
+			.lock()
+			.as_ref()
+			.map(|resources| Arc::clone(&resources.peer))
+			.ok_or_else(|| "Native live WebRTC peer has not started".to_owned())?;
+		let answer = RTCSessionDescription::answer(sdp)
+			.map_err(|error| format!("Codex returned an invalid live SDP answer: {error}"))?;
+		peer
+			.set_remote_description(answer)
+			.await
+			.map_err(|error| format!("Failed to install the live SDP answer: {error}"))
+	}
+
+	/// Wait until the `oai-events` data channel is open, failing on peer
+	/// failure, close, or timeout.
+	pub async fn wait_for_open(&self, timeout_ms: u32) -> VoiceResult<()> {
+		let mut signal_rx = self.signal_tx.subscribe();
+		let wait = async {
+			loop {
+				let signal = signal_rx.borrow().clone();
+				match signal {
+					PeerSignal::Open => return Ok(()),
+					PeerSignal::Failed(message) => return Err(message),
+					PeerSignal::Closed => {
+						return Err("Native live WebRTC peer closed before opening".to_owned());
+					},
+					PeerSignal::Connecting => {},
+				}
+				signal_rx
+					.changed()
+					.await
+					.map_err(|_| "Native live WebRTC peer stopped before opening".to_owned())?;
+			}
+		};
+		tokio::time::timeout(Duration::from_millis(u64::from(timeout_ms)), wait)
+			.await
+			.map_err(|_| "Timed out waiting for the live data channel to open".to_owned())?
+	}
+
+	/// Queue 16 kHz mono PCM for Opus transmission. Silently drops audio while
+	/// muted or when the bounded input queue is full.
+	pub fn push_audio(&self, samples: &[f32]) -> VoiceResult<()> {
+		if samples.is_empty() || self.muted.load(Ordering::Acquire) {
+			return Ok(());
+		}
+		let input_tx = self
+			.resources
+			.lock()
+			.as_ref()
+			.map(|resources| resources.input_tx.clone())
+			.ok_or_else(|| "Native live WebRTC peer has not started".to_owned())?;
+		let sample_count = samples.len().min(MAX_QUEUED_INPUT_SAMPLES);
+		let retained = &samples[samples.len() - sample_count..];
+		let queued = self
+			.queued_samples
+			.fetch_add(sample_count, Ordering::AcqRel);
+		if queued.saturating_add(sample_count) > MAX_QUEUED_INPUT_SAMPLES {
+			self
+				.queued_samples
+				.fetch_sub(sample_count, Ordering::AcqRel);
+			return Ok(());
+		}
+		if input_tx
+			.send(InputCommand::Audio(retained.to_vec()))
+			.is_err()
+		{
+			self
+				.queued_samples
+				.fetch_sub(sample_count, Ordering::AcqRel);
+			return Err("Native live audio input is closed".to_owned());
+		}
+		Ok(())
+	}
+
+	/// Enable or disable microphone transmission, discarding partial muted
+	/// frames.
+	pub fn set_muted(&self, muted: bool) -> VoiceResult<()> {
+		self.muted.store(muted, Ordering::Release);
+		let input_tx = self
+			.resources
+			.lock()
+			.as_ref()
+			.map(|resources| resources.input_tx.clone());
+		if let Some(input_tx) = input_tx {
+			input_tx
+				.send(InputCommand::Muted(muted))
+				.map_err(|_| "Native live audio input is closed".to_owned())?;
+		}
+		Ok(())
+	}
+
+	/// Whether close has begun; lets the adapter's `Drop` skip spawning a
+	/// redundant close task.
+	pub fn is_closing(&self) -> bool {
+		self.closing.load(Ordering::Acquire)
+	}
+
+	fn report_event(&self, payload: String) {
+		(self.callbacks.event)(payload);
+	}
+
+	fn report_level(&self, level: f64) {
+		(self.callbacks.level)(level.clamp(0.0, 1.0));
+	}
+
+	fn mark_open(&self) {
+		if !self.closing.load(Ordering::Acquire) {
+			self.signal_tx.send_replace(PeerSignal::Open);
+		}
+	}
+
+	fn report_failure(&self, message: String) {
+		if self.closing.load(Ordering::Acquire) || self.failure_reported.swap(true, Ordering::AcqRel)
+		{
+			return;
+		}
+		self
+			.signal_tx
+			.send_replace(PeerSignal::Failed(message.clone()));
+		(self.callbacks.failure)(message);
+	}
+
+	/// Close media, the data channel, the peer connection, and speaker
+	/// playback. Concurrent calls wait for the first closer to finish.
+	pub async fn close(&self) {
+		if self.closing.swap(true, Ordering::AcqRel) {
+			let mut signal_rx = self.signal_tx.subscribe();
+			while !matches!(*signal_rx.borrow(), PeerSignal::Closed) {
+				if signal_rx.changed().await.is_err() {
+					break;
+				}
+			}
+			return;
+		}
+
+		let resources = self.resources.lock().take();
+		if let Some(mut resources) = resources {
+			let _ = resources.input_tx.send(InputCommand::Close);
+			let _ = resources.peer.close().await;
+			let _ = resources.playback.stop();
+			let _ = tokio::time::timeout(CLOSE_TASK_TIMEOUT, resources.input_task).await;
+			resources.rtcp_task.abort();
+			let _ = resources.rtcp_task.await;
+			drop(resources.data_channel);
+		}
+		self.queued_samples.store(0, Ordering::Release);
+		self.signal_tx.send_replace(PeerSignal::Closed);
+	}
+}
+
+fn opus_capability() -> RTCRtpCodecCapability {
+	RTCRtpCodecCapability {
+		mime_type:     MIME_TYPE_OPUS.to_owned(),
+		clock_rate:    OPUS_CAPABILITY.clock_rate,
+		channels:      OPUS_CAPABILITY.channels,
+		sdp_fmtp_line: "minptime=10;useinbandfec=1".to_owned(),
+		rtcp_feedback: Vec::new(),
+	}
+}
+
+fn install_peer_callbacks(
+	peer: &Arc<RTCPeerConnection>,
+	core: Weak<LivePeerCore>,
+	playback_tx: PlaybackWriter,
+) {
+	let output_sender = Arc::new(Mutex::new(Some(playback_tx)));
+	let output_sender_for_track = Arc::clone(&output_sender);
+	let core_for_track = core.clone();
+	peer.on_track(Box::new(move |track, _receiver, _transceiver| {
+		let output_sender = output_sender_for_track.lock().take();
+		let core = core_for_track.clone();
+		Box::pin(async move {
+			if track.kind() != RTPCodecType::Audio {
+				return;
+			}
+			let Some(output_sender) = output_sender else {
+				if let Some(core) = core.upgrade() {
+					core.report_failure(
+						"Codex live returned more than one remote audio track".to_owned(),
+					);
+				}
+				return;
+			};
+			tokio::spawn(receive_output_audio(track, output_sender, core));
+		})
+	}));
+
+	let peer_for_state = Arc::downgrade(peer);
+	peer.on_peer_connection_state_change(Box::new(move |state| {
+		let core = core.clone();
+		let peer = peer_for_state.clone();
+		Box::pin(async move {
+			let Some(core) = core.upgrade() else {
+				return;
+			};
+			match state {
+				RTCPeerConnectionState::Failed => {
+					core.report_failure("Live WebRTC peer connection failed".to_owned());
+				},
+				RTCPeerConnectionState::Closed if !core.closing.load(Ordering::Acquire) => {
+					core.report_failure("Live WebRTC peer connection closed unexpectedly".to_owned());
+				},
+				RTCPeerConnectionState::Disconnected => {
+					tokio::time::sleep(DISCONNECT_GRACE).await;
+					if peer.upgrade().is_some_and(|peer| {
+						peer.connection_state() == RTCPeerConnectionState::Disconnected
+					}) {
+						core.report_failure("Live WebRTC peer connection disconnected".to_owned());
+					}
+				},
+				_ => {},
+			}
+		})
+	}));
+}
+
+fn install_data_channel_callbacks(data_channel: &Arc<RTCDataChannel>, core: Weak<LivePeerCore>) {
+	let core_for_open = core.clone();
+	data_channel.on_open(Box::new(move || {
+		Box::pin(async move {
+			if let Some(core) = core_for_open.upgrade() {
+				core.mark_open();
+			}
+		})
+	}));
+
+	let core_for_message = core.clone();
+	data_channel.on_message(Box::new(move |message: DataChannelMessage| {
+		let core = core_for_message.clone();
+		Box::pin(async move {
+			if !message.is_string {
+				return;
+			}
+			if let (Some(core), Ok(payload)) =
+				(core.upgrade(), String::from_utf8(message.data.to_vec()))
+			{
+				core.report_event(payload);
+			}
+		})
+	}));
+
+	let core_for_close = core.clone();
+	data_channel.on_close(Box::new(move || {
+		let core = core_for_close.clone();
+		Box::pin(async move {
+			if let Some(core) = core.upgrade() {
+				core.report_failure("Live data channel closed unexpectedly".to_owned());
+			}
+		})
+	}));
+
+	data_channel.on_error(Box::new(move |error| {
+		let core = core.clone();
+		Box::pin(async move {
+			if let Some(core) = core.upgrade() {
+				core.report_failure(format!("Live data channel failed: {error}"));
+			}
+		})
+	}));
+}
+
+async fn run_input_audio(
+	track: Arc<TrackLocalStaticSample>,
+	input_rx: flume::Receiver<InputCommand>,
+	core: Weak<LivePeerCore>,
+) {
+	let mut encoder = match Encoder::new(INPUT_SAMPLE_RATE, Channels::Mono, Application::Voip) {
+		Ok(encoder) => encoder,
+		Err(error) => {
+			if let Some(core) = core.upgrade() {
+				core.report_failure(format!("Failed to initialize the live Opus encoder: {error}"));
+			}
+			return;
+		},
+	};
+	if let Err(error) = encoder.set_inband_fec(true) {
+		if let Some(core) = core.upgrade() {
+			core.report_failure(format!("Failed to configure the live Opus encoder: {error}"));
+		}
+		return;
+	}
+
+	let mut muted = false;
+	let mut pending = Vec::with_capacity(INPUT_FRAME_SAMPLES * 2);
+	let mut encoded = [0u8; MAX_ENCODED_OPUS_BYTES];
+	let mut ticker = tokio::time::interval(INPUT_FRAME_DURATION);
+	ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Burst);
+	ticker.tick().await;
+	loop {
+		tokio::select! {
+			biased;
+			command = input_rx.recv_async() => {
+				let Ok(command) = command else {
+					break;
+				};
+				match command {
+					InputCommand::Audio(samples) => {
+						if let Some(core) = core.upgrade() {
+							core.queued_samples.fetch_sub(samples.len(), Ordering::AcqRel);
+						}
+						if muted {
+							continue;
+						}
+						if samples.len() >= MAX_QUEUED_INPUT_SAMPLES {
+							pending.clear();
+							pending.extend_from_slice(&samples[samples.len() - MAX_QUEUED_INPUT_SAMPLES..]);
+							continue;
+						}
+						let overflow = pending
+							.len()
+							.saturating_add(samples.len())
+							.saturating_sub(MAX_QUEUED_INPUT_SAMPLES);
+						if overflow > 0 {
+							pending.drain(..overflow);
+						}
+						pending.extend_from_slice(&samples);
+					},
+					InputCommand::Muted(next_muted) => {
+						muted = next_muted;
+						pending.clear();
+					},
+					InputCommand::Close => break,
+				}
+			},
+			_ = ticker.tick() => {
+				let mut frame = [0.0f32; INPUT_FRAME_SAMPLES];
+				if !muted {
+					let consumed = pending.len().min(INPUT_FRAME_SAMPLES);
+					frame[..consumed].copy_from_slice(&pending[..consumed]);
+					if consumed > 0 {
+						pending.copy_within(consumed.., 0);
+						pending.truncate(pending.len() - consumed);
+					}
+				}
+				let encoded_len = match encoder.encode_float(&frame, &mut encoded) {
+					Ok(encoded_len) => encoded_len,
+					Err(error) => {
+						if let Some(core) = core.upgrade() {
+							core.report_failure(format!("Failed to encode live microphone audio: {error}"));
+						}
+						return;
+					},
+				};
+				let sample = Sample {
+					data: Bytes::copy_from_slice(&encoded[..encoded_len]),
+					duration: INPUT_FRAME_DURATION,
+					..Default::default()
+				};
+				if let Err(error) = track.write_sample(&sample).await {
+					if let Some(core) = core.upgrade() {
+						core.report_failure(format!("Failed to send live microphone audio: {error}"));
+					}
+					return;
+				}
+			},
+		}
+	}
+}
+
+async fn drain_rtcp(sender: Arc<RTCRtpSender>) {
+	while sender.read_rtcp().await.is_ok() {}
+}
+
+async fn receive_output_audio(
+	track: Arc<TrackRemote>,
+	playback_tx: PlaybackWriter,
+	core: Weak<LivePeerCore>,
+) {
+	if !track
+		.codec()
+		.capability
+		.mime_type
+		.eq_ignore_ascii_case(MIME_TYPE_OPUS)
+	{
+		if let Some(core) = core.upgrade() {
+			core.report_failure(format!(
+				"Codex live negotiated unsupported audio codec {}",
+				track.codec().capability.mime_type
+			));
+		}
+		return;
+	}
+	let mut decoder = match Decoder::new(OUTPUT_SAMPLE_RATE, Channels::Mono) {
+		Ok(decoder) => decoder,
+		Err(error) => {
+			if let Some(core) = core.upgrade() {
+				core.report_failure(format!("Failed to initialize the live Opus decoder: {error}"));
+			}
+			return;
+		},
+	};
+	let mut decoded = vec![0.0f32; MAX_DECODED_OPUS_SAMPLES].into_boxed_slice();
+	let mut expected_sequence: Option<u16> = None;
+	let mut level = OutputLevel::default();
+
+	loop {
+		let packet = match track.read_rtp().await {
+			Ok((packet, _attributes)) => packet,
+			Err(error) => {
+				if let Some(core) = core.upgrade()
+					&& !core.closing.load(Ordering::Acquire)
+				{
+					core.report_failure(format!("Live remote audio track failed: {error}"));
+				}
+				return;
+			},
+		};
+		let sequence = packet.header.sequence_number;
+		if let Some(expected) = expected_sequence {
+			let gap = sequence.wrapping_sub(expected);
+			if gap >= u16::MAX / 2 {
+				continue;
+			}
+			if gap > 0 {
+				for _ in 1..gap.min(5) {
+					if let Ok(samples) =
+						decoder.decode_float(&[], &mut decoded[..OUTPUT_FRAME_SAMPLES], false)
+					{
+						if !write_output(&playback_tx, &decoded[..samples], &core) {
+							return;
+						}
+						level.observe(&decoded[..samples], &core);
+					}
+				}
+				if let Ok(samples) = decoder.decode_float(&packet.payload, &mut decoded, true) {
+					if !write_output(&playback_tx, &decoded[..samples], &core) {
+						return;
+					}
+					level.observe(&decoded[..samples], &core);
+				}
+			}
+		}
+		expected_sequence = Some(sequence.wrapping_add(1));
+		match decoder.decode_float(&packet.payload, &mut decoded, false) {
+			Ok(samples) => {
+				if !write_output(&playback_tx, &decoded[..samples], &core) {
+					return;
+				}
+				level.observe(&decoded[..samples], &core);
+			},
+			Err(error) => {
+				if let Some(core) = core.upgrade() {
+					core.report_failure(format!("Failed to decode live speaker audio: {error}"));
+				}
+				return;
+			},
+		}
+	}
+}
+
+fn write_output(playback_tx: &PlaybackWriter, samples: &[f32], core: &Weak<LivePeerCore>) -> bool {
+	match playback_tx.write(samples) {
+		Ok(()) => true,
+		Err(error) => {
+			if let Some(core) = core.upgrade()
+				&& !core.closing.load(Ordering::Acquire)
+			{
+				core.report_failure(format!("Live speaker playback failed: {error}"));
+			}
+			false
+		},
+	}
+}
+
+#[derive(Default)]
+struct OutputLevel {
+	sum_squares: f64,
+	samples:     usize,
+}
+
+impl OutputLevel {
+	fn observe(&mut self, decoded: &[f32], core: &Weak<LivePeerCore>) {
+		let mut offset = 0;
+		while offset < decoded.len() {
+			let take = (OUTPUT_LEVEL_SAMPLES - self.samples).min(decoded.len() - offset);
+			for &sample in &decoded[offset..offset + take] {
+				let sample = f64::from(sample);
+				self.sum_squares = sample.mul_add(sample, self.sum_squares);
+			}
+			self.samples += take;
+			offset += take;
+			if self.samples == OUTPUT_LEVEL_SAMPLES {
+				if let Some(core) = core.upgrade() {
+					core.report_level((self.sum_squares / self.samples as f64).sqrt());
+				}
+				self.sum_squares = 0.0;
+				self.samples = 0;
+			}
+		}
+	}
+}

docs/natives-build-release-debugging.md

@@ -146,7 +146,7 @@ On non-PR events both jobs run on `omp-kata` pods against the cluster remote cac
 bazelisk --bazelrc="$rc" test //crates/...                 # full Rust suite
 # clippy scope mirrors `cargo clippy --workspace` (libraries only), split by
 # lint policy via a query kind filter:
-bazelisk query "kind('rust_library|rust_shared_library', //crates/pi-ast/... + //crates/pi-iso/... + //crates/pi-natives/... + //crates/pi-shell/... + //crates/pi-walker/...)" \
+bazelisk query "kind('rust_library|rust_shared_library', //crates/pi-ast/... + //crates/pi-iso/... + //crates/pi-natives/... + //crates/pi-shell/... + //crates/pi-voice/... + //crates/pi-walker/...)" \
   | xargs bazelisk --bazelrc="$rc" build --config=clippy-strict --
 bazelisk query "kind('rust_library|rust_shared_library', //crates/... - (…strict set…) - //crates/vendor/brush-core/... - //crates/vendor/brush-builtins/...)" \
   | xargs bazelisk --bazelrc="$rc" build --config=clippy --
@@ -179,9 +179,9 @@ Hosted disk caches use `bazel-disk-v3-<scope>-<os>-<arch>-<config-hash>-<source-
 
 `.github/actions/bazel-natives` is the direct builder: `bazel-cache` → `OMP_BAZEL_RC=<rc> bun scripts/bazel-natives.ts <targets> --dest <dest>`, followed by a disk-cache save after a hosted miss. `.github/actions/native-artifacts` is the no-build consumer: download `native-addons` → run the same driver with `--source`.
 
-### `release_binary`
+### Release binary builds and publishing
 
-Linux and Windows release matrices install addons from the `native_addons` workflow artifact. Darwin artifacts cannot be cross-built on Linux, so each macOS matrix builds only its own architecture through `bazel-natives` with scope `release-<target_id>`, then `bun run ci:release:build-binaries` embeds and compiles the executable.
+Binary builds are build-only and run in parallel with the test fan-out. `release_binary` (Linux + Windows matrices) needs only `native_addons`, whose workflow artifact supplies their addons. `release_binary_darwin` needs only `release_metadata` and starts the moment a release run is detected: darwin artifacts cannot be cross-built on Linux, so each macOS leg builds its own architecture through `bazel-natives` with scope `release-<target_id>` (seeded near HEAD by the warm workflow — normally just the version-bump delta), then `bun run ci:release:build-binaries` embeds and compiles the executable. Publishing is held behind `release_gate` (the aggregate of every validation job): `release_native_leaves` downloads all built addons and publishes the five `@oh-my-pi/pi-natives-<tag>` leaves from one linux runner, and the GitHub release / verify / core npm chain runs beside it.
 
 ## Debugging playbook
 

packages/natives/CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## [Unreleased]
 
+### Changed
+
+- Split the native voice engine (miniaudio capture/playback, WebRTC peer, Opus media) out of the `pi-natives` addon crate into a napi-free `pi-voice` rlib. The addon keeps thin `#[napi]` adapters, so the JS API is unchanged; the webrtc/opus/miniaudio dependency graph now compiles once into the library and no longer rebuilds with the addon leaf (which recompiles every release via its version-sentinel edit).
+- Release binaries now build in parallel with the test fan-out; npm leaf publishing moved to a dedicated post-validation job (`release_native_leaves`), and darwin release bazel caches are pre-warmed on native-affecting main pushes — cutting release wall time from the previous serialized tests → cold darwin build pipeline.
+
 ## [17.1.8] - 2026-07-28
 
 ### Fixed