feat(moq-audio)!: codecs return zero or more outputs and flush

cpp/obs/src/moq-audio-encoder.cpp

@@ -114,7 +114,23 @@ bool Encode(void *data, struct encoder_frame *frame, struct encoder_packet *pack
 		LOG_ERROR("MoQ Opus encode failed: %s", encoded.error().to_string().c_str());
 		return false;
 	}
-	enc->packet = std::move(encoded->data);
+	// libopus readies each frame's packet as soon as it is submitted, and OBS
+	// takes at most one packet per call.
+	auto received = enc->codec->receive();
+	if (!received) {
+		LOG_ERROR("MoQ Opus encode failed: %s", received.error().to_string().c_str());
+		return false;
+	}
+	if (!*received) {
+		LOG_ERROR("MoQ Opus encoder held a frame's packet");
+		return false;
+	}
+	enc->packet = std::move((*received)->data);
+	auto extra = enc->codec->receive();
+	if (!extra || *extra) {
+		LOG_ERROR("MoQ Opus encoder readied more than one packet for one frame");
+		return false;
+	}
 
 	// One packet per frame, stamped with its first input sample: the OpusHead's
 	// pre-skip already tells the decoder to trim the codec's lookahead.

dart/moq_ffi/lib/src/moq.dart

@@ -9427,9 +9427,11 @@ class FfiConverterMoqAudioConsumer {
 }
 
 abstract class MoqAudioEncoderInterface {
-  MoqAudioPacket encode({required List<double> pcm});
+  void encode({required List<double> pcm});
   MoqAudioEncoderFinished finish({required List<double> pcm});
+  void flush();
   int frameSize();
+  MoqAudioPacket? receive();
   void setBitrate({required int bitrate});
   MoqAudioInit trackInit();
 }
@@ -9481,16 +9483,14 @@ class MoqAudioEncoder implements MoqAudioEncoderInterface {
     }
   }
 
-  MoqAudioPacket encode({required List<double> pcm}) {
-    return rustCallWithLifter(
-      (status) => uniffi_moq_ffi_fn_method_moqaudioencoder_encode(
+  void encode({required List<double> pcm}) {
+    return rustCall((status) {
+      uniffi_moq_ffi_fn_method_moqaudioencoder_encode(
         uniffiClonePointer(),
         FfiConverterSequenceDouble32.lower(pcm),
         status,
-      ),
-      FfiConverterMoqAudioPacket.lift,
-      moqExceptionErrorHandler,
-    );
+      );
+    }, moqExceptionErrorHandler);
   }
 
   MoqAudioEncoderFinished finish({required List<double> pcm}) {
@@ -9505,6 +9505,15 @@ class MoqAudioEncoder implements MoqAudioEncoderInterface {
     );
   }
 
+  void flush() {
+    return rustCall((status) {
+      uniffi_moq_ffi_fn_method_moqaudioencoder_flush(
+        uniffiClonePointer(),
+        status,
+      );
+    }, moqExceptionErrorHandler);
+  }
+
   int frameSize() {
     return rustCallWithLifter(
       (status) => uniffi_moq_ffi_fn_method_moqaudioencoder_frame_size(
@@ -9516,6 +9525,17 @@ class MoqAudioEncoder implements MoqAudioEncoderInterface {
     );
   }
 
+  MoqAudioPacket? receive() {
+    return rustCallWithLifter(
+      (status) => uniffi_moq_ffi_fn_method_moqaudioencoder_receive(
+        uniffiClonePointer(),
+        status,
+      ),
+      FfiConverterOptionalMoqAudioPacket.lift,
+      moqExceptionErrorHandler,
+    );
+  }
+
   void setBitrate({required int bitrate}) {
     return rustCall((status) {
       uniffi_moq_ffi_fn_method_moqaudioencoder_set_bitrate(
@@ -17327,6 +17347,53 @@ class FfiConverterOptionalMoqAnnounceEvent {
   }
 }
 
+class FfiConverterOptionalMoqAudioPacket {
+  static MoqAudioPacket? lift(RustBuffer buf) {
+    return FfiConverterOptionalMoqAudioPacket.read(buf.asUint8List()).value;
+  }
+
+  static LiftRetVal<MoqAudioPacket?> read(Uint8List buf) {
+    if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) {
+      return LiftRetVal(null, 1);
+    }
+    final result = FfiConverterMoqAudioPacket.read(
+      Uint8List.view(buf.buffer, buf.offsetInBytes + 1),
+    );
+    return LiftRetVal<MoqAudioPacket?>(result.value, result.bytesRead + 1);
+  }
+
+  static int allocationSize([MoqAudioPacket? value]) {
+    if (value == null) {
+      return 1;
+    }
+    return FfiConverterMoqAudioPacket.allocationSize(value) + 1;
+  }
+
+  static RustBuffer lower(MoqAudioPacket? value) {
+    if (value == null) {
+      return toRustBuffer(Uint8List.fromList([0]));
+    }
+    final buf = Uint8List(
+      FfiConverterOptionalMoqAudioPacket.allocationSize(value),
+    );
+    FfiConverterOptionalMoqAudioPacket.write(value, buf);
+    return toRustBuffer(buf);
+  }
+
+  static int write(MoqAudioPacket? value, Uint8List buf) {
+    if (value == null) {
+      buf[0] = 0;
+      return 1;
+    }
+    buf[0] = 1;
+    return FfiConverterMoqAudioPacket.write(
+          value,
+          Uint8List.view(buf.buffer, buf.offsetInBytes + 1),
+        ) +
+        1;
+  }
+}
+
 class FfiConverterOptionalMoqAudioRead {
   static MoqAudioRead? lift(RustBuffer buf) {
     return FfiConverterOptionalMoqAudioRead.read(buf.asUint8List()).value;
@@ -19759,10 +19826,10 @@ external Pointer<Void> uniffi_moq_ffi_fn_constructor_moqaudioencoder_new(
   Pointer<RustCallStatus> uniffiStatus,
 );
 
-@Native<
-  RustBuffer Function(Pointer<Void>, RustBuffer, Pointer<RustCallStatus>)
->(assetId: _uniffiAssetId)
-external RustBuffer uniffi_moq_ffi_fn_method_moqaudioencoder_encode(
+@Native<Void Function(Pointer<Void>, RustBuffer, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external void uniffi_moq_ffi_fn_method_moqaudioencoder_encode(
   Pointer<Void> ptr,
   RustBuffer pcm,
   Pointer<RustCallStatus> uniffiStatus,
@@ -19777,6 +19844,14 @@ external RustBuffer uniffi_moq_ffi_fn_method_moqaudioencoder_finish(
   Pointer<RustCallStatus> uniffiStatus,
 );
 
+@Native<Void Function(Pointer<Void>, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external void uniffi_moq_ffi_fn_method_moqaudioencoder_flush(
+  Pointer<Void> ptr,
+  Pointer<RustCallStatus> uniffiStatus,
+);
+
 @Native<Uint64 Function(Pointer<Void>, Pointer<RustCallStatus>)>(
   assetId: _uniffiAssetId,
 )
@@ -19785,6 +19860,14 @@ external int uniffi_moq_ffi_fn_method_moqaudioencoder_frame_size(
   Pointer<RustCallStatus> uniffiStatus,
 );
 
+@Native<RustBuffer Function(Pointer<Void>, Pointer<RustCallStatus>)>(
+  assetId: _uniffiAssetId,
+)
+external RustBuffer uniffi_moq_ffi_fn_method_moqaudioencoder_receive(
+  Pointer<Void> ptr,
+  Pointer<RustCallStatus> uniffiStatus,
+);
+
 @Native<Void Function(Pointer<Void>, Uint64, Pointer<RustCallStatus>)>(
   assetId: _uniffiAssetId,
 )
@@ -23282,9 +23365,15 @@ external int uniffi_moq_ffi_checksum_method_moqaudioencoder_encode();
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqaudioencoder_finish();
 
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqaudioencoder_flush();
+
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqaudioencoder_frame_size();
 
+@Native<Uint16 Function()>(assetId: _uniffiAssetId)
+external int uniffi_moq_ffi_checksum_method_moqaudioencoder_receive();
+
 @Native<Uint16 Function()>(assetId: _uniffiAssetId)
 external int uniffi_moq_ffi_checksum_method_moqaudioencoder_set_bitrate();
 
@@ -24146,15 +24235,21 @@ void _checkApiChecksums() {
   if (uniffi_moq_ffi_checksum_method_moqaudioconsumer_next() != 7650) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
-  if (uniffi_moq_ffi_checksum_method_moqaudioencoder_encode() != 56589) {
+  if (uniffi_moq_ffi_checksum_method_moqaudioencoder_encode() != 12465) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
   if (uniffi_moq_ffi_checksum_method_moqaudioencoder_finish() != 24341) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
+  if (uniffi_moq_ffi_checksum_method_moqaudioencoder_flush() != 43516) {
+    throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
+  }
   if (uniffi_moq_ffi_checksum_method_moqaudioencoder_frame_size() != 37264) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }
+  if (uniffi_moq_ffi_checksum_method_moqaudioencoder_receive() != 46467) {
+    throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
+  }
   if (uniffi_moq_ffi_checksum_method_moqaudioencoder_set_bitrate() != 10667) {
     throw UniffiInternalError.panicked("UniFFI API checksum mismatch");
   }

go/wrapper/audio/audio.go

@@ -207,8 +207,14 @@ func (e *Encoder) TrackInit() (media.AudioInit, error) { return e.inner.TrackIni
 // SetBitrate retunes the live encoder in bits per second.
 func (e *Encoder) SetBitrate(value uint64) error { return e.inner.SetBitrate(value) }
 
-// Encode encodes one codec-sized interleaved f32 frame.
-func (e *Encoder) Encode(pcm []float32) (Packet, error) { return e.inner.Encode(pcm) }
+// Encode submits one codec-sized interleaved f32 frame, whose packet Receive returns.
+func (e *Encoder) Encode(pcm []float32) error { return e.inner.Encode(pcm) }
+
+// Receive pops the oldest ready packet, or nil while the codec holds the rest; call it until nil.
+func (e *Encoder) Receive() (*Packet, error) { return e.inner.Receive() }
+
+// Flush marks the end of input, so Receive returns every packet the codec still holds, without closing the encoder.
+func (e *Encoder) Flush() error { return e.inner.Flush() }
 
 // Finish pads an optional partial frame, drains the codec, and closes the encoder.
 func (e *Encoder) Finish(pcm []float32) (EncoderFinish, error) { return e.inner.Finish(pcm) }

py/moq-rs/moq/audio.py

@@ -223,9 +223,17 @@ def set_bitrate(self, bitrate: int) -> None:
         """Retune the live encoder in bits per second."""
         self._inner.set_bitrate(bitrate)
 
-    def encode(self, pcm: list[float]) -> Packet:
-        """Encode one full PCM frame."""
-        return self._inner.encode(pcm)
+    def encode(self, pcm: list[float]) -> None:
+        """Submit one full PCM frame, whose packet ``receive`` returns."""
+        self._inner.encode(pcm)
+
+    def receive(self) -> Packet | None:
+        """Pop the oldest ready packet, or ``None`` while the codec holds the rest; call until ``None``."""
+        return self._inner.receive()
+
+    def flush(self) -> None:
+        """Mark the end of input, so ``receive`` returns every packet the codec still holds."""
+        self._inner.flush()
 
     def finish(self, pcm: list[float] | None = None) -> EncoderFinish:
         """Pad an optional partial frame and drain the codec."""

py/moq-rs/tests/test_local.py

@@ -1548,9 +1548,13 @@ async def test_read_frame_keeps_group_across_cancelled_call():
 def test_audio_encoder_returns_namespaced_records():
     """The codec-only encoder's results are named under ``moq.audio``."""
     encoder = moq.audio.Encoder(moq.audio.EncoderOutput(codec=moq.AudioCodec.opus(), sample_rate=48000, channels=1))
-    packet = encoder.encode([0.0] * encoder.frame_size)
+    encoder.encode([0.0] * encoder.frame_size)
+    packet = encoder.receive()
     assert isinstance(packet, moq.audio.Packet)
     assert packet.data
+    assert encoder.receive() is None
+    encoder.flush()
+    assert encoder.receive() is None
     finish = encoder.finish()
     assert isinstance(finish, moq.audio.EncoderFinish)
 
@@ -1566,7 +1570,9 @@ def test_audio_encoder_frame_init_and_bitrate():
     assert init.format == moq.media.AudioFormat.OPUS
     assert init.data[:8] == b"OpusHead"
     encoder.set_bitrate(32_000)
-    assert encoder.encode([0.0] * encoder.frame_size).data
+    encoder.encode([0.0] * encoder.frame_size)
+    packet = encoder.receive()
+    assert packet is not None and packet.data
 
     with pytest.raises(moq.Error.Audio):
         moq.audio.Encoder(replace(output, frame_duration=timedelta(milliseconds=15)))

quest/m2/README.md

@@ -197,13 +197,13 @@ quests that require them.
 - [Unbounded groups](/quest/m2/unbounded-groups/README.md) - groups and frames lose their size caps; a group keeps a 32 MB tail, and readers behind it replay through paced flights or refresh
 - [moq-quic drops inherited surfaces with no consumer](/quest/m2/moq-quic-unused-surface.md) - unused accept, connection, and stream-id surfaces leave moq-quic's public API before the cut
 - [No sentinel values in public API](/quest/m2/no-sentinels.md) - receive windows, cache durations, reconnect timeouts, and the FFI test clock use `Option` instead of a huge value or zero
+- [moq-video codecs split submission from receive](/quest/m2/video-codec-receive.md) - video encoders and decoders submit, then `receive` one output, the shape moq-audio took, so no call allocates a `Vec` of outputs
 - [Connection state carries each phase's data](/quest/m2/connection-state-phases.md) - moq-tokio's `Observation`, moq-ffi's `MoqConnectionState`, and moq-c's `moq_connection_state` each hold one phase per value, so no field is readable in a phase where it means nothing
 - [moq-video takes devices, not environment variables](/quest/m2/video-devices.md) - VA-API and V4L2 device paths are config fields, and `moq_video::ENV` never ships
 - [Gateways share one ingest config and one rate cap](/quest/m2/gateway-ingest.md) - `moq_mux::Ingest` replaces 18 per-gateway setters, and the live rate cap moves to `moq_net::bandwidth`
 - [JS time values are types, not numbers](/quest/m2/js-time-types.md) - every public JS duration and instant is a `Clock.Duration` or `Clock.Instant` object, so a unit mistake fails to compile
 - [Small API fixes before the cut](/quest/m2/api-shape/README.md) - one path and one spelling per item, names by role, and no dead or wrong-role public surface across net, transport, media, native, apps, JS, and bindings
 - [Public constructors spawn nothing, or say why](/quest/m2/spawn-constructors.md) - each public library entry point that spawns returns a driver its caller polls, or is listed as kept
-- [moq-audio codecs return zero or more outputs](/quest/m2/audio-codec-seam.md) - `encode` and `decode` return any number of outputs and gain `flush`, as moq-video's do
 - [Plan: recovery boundaries](/quest/m2/intra-refresh-boundary.md) - the catalog's new `warmup` contract is settled before the cut publishes it
 - [MoQ on rind](/quest/m1/rind/README.md) - every MoQ crate takes its crypto from rind, a pure-Rust graviola fork, so no build compiles C crypto (held in m1 by an open claim)
 - [Finishing a producer returns a Finished handle](/quest/m2/finished-handle.md) - `finish(self)` consumes moq-net's and the media crates' producers, so appending after finish doesn't compile

quest/m2/video-codec-receive.md

@@ -0,0 +1,17 @@
+# [L] moq-video codecs split submission from receive
+
+## Goal
+
+moq-video's encoders and decoders take the same send/receive shape moq-audio took in [moq-dev/moq#8270](https://github.com/moq-dev/moq/pull/8270): `encode`/`decode` only submit and return `Result<()>`, `receive` pops one ready output (`Ok(None)` when nothing is ready), and `flush` marks the end of input so `receive` drains what the codec held.
+No call allocates a `Vec` just to hand back its outputs.
+
+## Plan
+
+- Decided 2026-11-08 (maintainer, on #8270): the FFmpeg/MediaCodec split, rejecting a caller-owned `&mut Vec` buffer and keeping `Vec`. moq-video still returns `Vec` from its backend traits, `encode::Sink`, and the decoder front end, so the two crates disagree until this lands.
+- Apply it to every backend (VA-API, NVDEC, Media Foundation, VideoToolbox, MediaCodec, openh264, and the rest under `rs/moq-video/src/{encode,decode}/backend/`), the front ends, and their tests. A backend that readies output synchronously holds it in a reused `VecDeque`.
+- Mirror it through moq-ffi (`MoqVideoEncoder.encode`/`flush`/`finish`), its Python, Swift, Go, Dart, and Kotlin wrappers, moq-c (`moq_video_encoder_encode`/`flush`, with a `moq_video_encoder_receive` like `moq_audio_encoder_receive`), and the OBS video encoder.
+- Benchmark the encode and decode paths base vs branch; allocations should drop or match.
+
+## Related
+
+- [moq-dev/moq#8270](https://github.com/moq-dev/moq/pull/8270) - the moq-audio change whose shape this copies

quest/m5/audio-decode-mediacodec.md

@@ -28,7 +28,7 @@ behind a new optional audio `mediacodec` feature and the decode seam, on `target
 - The binding ships in the moq-ffi Android slice, which is how Kotlin and Dart
   reach it.
 
-Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): [audio-codec-seam](/quest/m2/audio-codec-seam.md) makes moq-audio's codecs return zero or more outputs with `flush` before the cut, so this backend adds no break.
+Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): moq-audio's codecs return zero or more outputs with `flush` before the cut, so this backend adds no break. `decode::Consumer` still places each packet by the one output it returns and refuses any other count, so if this backend holds or splits output, it also teaches the consumer to place that output.
 
 ## Related
 

quest/m5/audio-decode-mediafoundation.md

@@ -22,7 +22,7 @@ ones. Behind the decode seam as the first candidate on `target_os =
 - Verification runs on a Windows host; the per-PR CI only compiles the
   platform code (`just rs windows`).
 
-Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): [audio-codec-seam](/quest/m2/audio-codec-seam.md) makes moq-audio's codecs return zero or more outputs with `flush` before the cut, so this backend adds no break.
+Changed 2026-11-07 (maintainer, /quest-plan from the API sweep of open PRs and remaining quests): moq-audio's codecs return zero or more outputs with `flush` before the cut, so this backend adds no break. `decode::Consumer` still places each packet by the one output it returns and refuses any other count, so if this backend holds or splits output, it also teaches the consumer to place that output.
 
 ## Related
 

quest/m5/audio-encode-audiotoolbox.md

@@ -30,13 +30,12 @@ the encode seam as the platform candidate on macOS and iOS.
   the deficit instead, including the terminal end (found in #4081 review).
 
 This backend implements the zero-or-more packet return and `flush` that
-[audio-codec-seam](/quest/m2/audio-codec-seam.md) lands in m2 before the cut
+moq-audio's encoder took in m2 before the cut
 (decided 2026-11-07, maintainer, /quest-plan from the API sweep of open PRs
 and remaining quests), so it adds no break and waits on no other encoder backend; its round-trip test still needs its platform decoder.
 
 ## Required
 
-- [moq-audio codecs return zero or more outputs](/quest/m2/audio-codec-seam.md) - the encoder API this backend implements
 - [AudioToolbox decode](/quest/m3/audio-decode-audiotoolbox.md) - the round-trip regression decodes through it
 
 ## Related