Add duplex stream API contract

CHANGELOG.md

@@ -7,6 +7,11 @@ public APIs may still change while the backend design stabilizes.
 
 ## [Unreleased]
 
+### Added
+
+- Added the public `DuplexStream` API contract for backend-level full-duplex
+  capture and playback.
+
 ## [0.2.0a3] - 2026-06-12
 
 ### Changed

README.md

@@ -53,6 +53,9 @@ Use blocking helpers when callers need complete buffer transfer:
 - `InputStream.read_exactly(frame_count, timeout=None)`: wait until exactly the
   requested number of frames has been captured
 
+Full-duplex capture/playback is exposed as `DuplexStream`. The public API is in
+place, but native backend implementations are still under development.
+
 Lifecycle semantics:
 
 - `stop()`: stop playback without discarding queued frames

docs/architecture.md

@@ -39,6 +39,11 @@ the output ring can hold, and `read()` returns currently available captured
 frames. Blocking helpers (`write_all()` and `read_exactly()`) are Python-level
 conveniences layered on top of those primitives.
 
+Full-duplex support is modeled as a backend-level `DuplexStream`, not as a
+Python wrapper around one `OutputStream` and one `InputStream`. Backends should
+use a single native duplex callback where available so capture and playback share
+one scheduling clock.
+
 Stream statistics distinguish queue state from hardware behavior. `queued_frames`
 and `queued_latency` describe the native ring buffer. `hardware_latency`
 describes backend-reported device latency when available. `buffer_size`

src/tachyaudio/__init__.py

@@ -3,13 +3,25 @@
 from tachyaudio._backend import get_backend, list_devices, set_backend
 from tachyaudio._device import DeviceInfo, DeviceKind
 from tachyaudio._errors import BackendUnavailable, StreamClosed, TachyAudioError
-from tachyaudio._stream import InputStream, OutputStream, StreamConfig, StreamStats, play
+from tachyaudio._stream import (
+    DuplexStream,
+    DuplexStreamConfig,
+    DuplexStreamStats,
+    InputStream,
+    OutputStream,
+    StreamConfig,
+    StreamStats,
+    play,
+)
 from tachyaudio._version import __version__
 
 __all__ = [
     "BackendUnavailable",
     "DeviceInfo",
     "DeviceKind",
+    "DuplexStream",
+    "DuplexStreamConfig",
+    "DuplexStreamStats",
     "InputStream",
     "OutputStream",
     "StreamClosed",

src/tachyaudio/_backend.py

@@ -42,6 +42,24 @@ def read(self, frame_count: int) -> memoryview: ...
     def stats(self) -> object: ...
 
 
+class DuplexStreamHandle(Protocol):
+    """Backend-owned full-duplex stream handle."""
+
+    def start(self) -> None: ...
+
+    def stop(self) -> None: ...
+
+    def flush(self) -> None: ...
+
+    def close(self) -> None: ...
+
+    def write(self, frames: Any) -> int: ...
+
+    def read(self, frame_count: int) -> memoryview: ...
+
+    def stats(self) -> object: ...
+
+
 class AudioBackend(Protocol):
     """Protocol implemented by concrete audio backends."""
 
@@ -53,6 +71,8 @@ def open_output_stream(self, config: object) -> OutputStreamHandle: ...
 
     def open_input_stream(self, config: object) -> InputStreamHandle: ...
 
+    def open_duplex_stream(self, config: object) -> DuplexStreamHandle: ...
+
 
 class _UnavailableBackend:
     name = "unavailable"
@@ -70,6 +90,11 @@ def open_input_stream(self, config: object) -> InputStreamHandle:
             "no tachyaudio backend is available yet; install or build the native backend"
         )
 
+    def open_duplex_stream(self, config: object) -> DuplexStreamHandle:
+        raise BackendUnavailable(
+            "no tachyaudio backend is available yet; install or build the native backend"
+        )
+
 
 def _load_default_backend() -> AudioBackend:
     try:

src/tachyaudio/_native_backend.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+from typing import NoReturn
+
 from tachyaudio import _native
 from tachyaudio._device import DeviceInfo, DeviceKind
+from tachyaudio._errors import BackendUnavailable
 
 
 class NativeOutputStream:
@@ -121,3 +124,7 @@ def open_output_stream(self, config: object) -> NativeOutputStream:
 
     def open_input_stream(self, config: object) -> NativeInputStream:
         return NativeInputStream(config)
+
+    def open_duplex_stream(self, config: object) -> NoReturn:
+        del config
+        raise BackendUnavailable("native duplex streams are not implemented yet")