feat(#4023): add adjustable local audio bridge gain controls

src/rigplane/audio/bridge.py

@@ -152,6 +152,24 @@ def _pcm16le_samples(pcm: bytes) -> array[int]:
     return samples
 
 
+def _apply_pcm_gain_db(pcm: bytes, gain_db: float) -> bytes:
+    """Apply gain to s16le PCM, saturating samples at the int16 limits."""
+    if gain_db == 0.0:
+        return pcm
+    if not math.isfinite(gain_db):
+        raise ValueError("Audio gain must be a finite dB value")
+
+    # A factor of 32768 already clips every nonzero int16 sample.
+    scale = 10.0 ** (min(gain_db, 20.0 * math.log10(32768)) / 20.0)
+    samples = _pcm16le_samples(pcm)
+    for index, sample in enumerate(samples):
+        scaled = round(sample * scale)
+        samples[index] = max(-32768, min(32767, scaled))
+    if sys.byteorder != "little":
+        samples.byteswap()
+    return samples.tobytes()
+
+
 def _downmix_stereo_to_mono(pcm: bytes) -> bytes:
     """Downmix L+R interleaved s16le → mono s16le via average.
 
@@ -288,6 +306,8 @@ class AudioBridge:
         sample_rate: PCM sample rate (default 48000).
         channels: Number of audio channels (default 1, mono).
         frame_ms: PCM frame duration in milliseconds (default 20).
+        input_gain_db: Gain applied to device-to-radio PCM (default 0 dB).
+        output_gain_db: Gain applied to radio-to-device PCM (default 0 dB).
         tx_enabled: Whether to bridge TX audio (device → radio). Default True.
         tx_executor: Accepted but ignored (deprecated) — the backend owns
             threading. Kept only so existing callers passing it keep working.
@@ -317,6 +337,8 @@ def __init__(
         sample_rate: int = SAMPLE_RATE,
         channels: int = CHANNELS,
         frame_ms: int = FRAME_MS,
+        input_gain_db: float = 0.0,
+        output_gain_db: float = 0.0,
         tx_enabled: bool = True,
         tx_executor: Executor | None = None,
         label: str = "rigplane",
@@ -336,6 +358,12 @@ def __init__(
         self._sample_rate = sample_rate
         self._channels = channels
         self._frame_ms = frame_ms
+        self._input_gain_db = 0.0
+        self._output_gain_db = 0.0
+        self.set_gains(
+            input_gain_db=input_gain_db,
+            output_gain_db=output_gain_db,
+        )
         self._tx_enabled = tx_enabled
         self._tx_started = False
         self._backend: AudioBackend = backend or PortAudioBackend()
@@ -422,6 +450,29 @@ def label(self) -> str:
     def running(self) -> bool:
         return self._running
 
+    @property
+    def input_gain_db(self) -> float:
+        return self._input_gain_db
+
+    @property
+    def output_gain_db(self) -> float:
+        return self._output_gain_db
+
+    def set_gains(
+        self,
+        *,
+        input_gain_db: float | None = None,
+        output_gain_db: float | None = None,
+    ) -> None:
+        """Update local bridge gains without changing shared Web UI audio."""
+        for value in (input_gain_db, output_gain_db):
+            if value is not None and not math.isfinite(value):
+                raise ValueError("Audio gain must be a finite dB value")
+        if input_gain_db is not None:
+            self._input_gain_db = input_gain_db
+        if output_gain_db is not None:
+            self._output_gain_db = output_gain_db
+
     @property
     def bridge_state(self) -> BridgeState:
         return self._bridge_state
@@ -1189,15 +1240,16 @@ async def _rx_loop(self) -> None:
                         if len(self._rx_latency_samples) > 100:
                             self._rx_latency_samples.pop(0)
                     self._last_rx_time = now
+                    out_data = pcm_data
+                    if self._input_channels == 2:
+                        out_data = _downmix_stereo_to_mono(out_data)
+                    out_data = _apply_pcm_gain_db(out_data, self._output_gain_db)
                     self._rx_frames += 1
-                    self._last_rx_level_dbfs = _rms_dbfs(pcm_data)
+                    self._last_rx_level_dbfs = _rms_dbfs(out_data)
                     if self._rx_frames % 50 == 0:
                         self._emit_metrics()
                     playback = self._playback_stream
                     if playback and playback.running:
-                        out_data = pcm_data
-                        if self._input_channels == 2:
-                            out_data = _downmix_stereo_to_mono(out_data)
                         await playback.write(out_data)
                 except OSError:
                     raise  # device-level error → outer handler → reconnect
@@ -1245,6 +1297,7 @@ async def _tx_loop(self) -> None:
                     self._drop_queued_tx()
                     continue
 
+                pcm_bytes = _apply_pcm_gain_db(pcm_bytes, self._input_gain_db)
                 samples = _pcm16le_samples(pcm_bytes)
                 peak = max((abs(sample) for sample in samples), default=0)
                 if peak < silence_threshold:

src/rigplane/cli/__init__.py

@@ -579,6 +579,16 @@ def _apply_managed_runtime_defaults(args: argparse.Namespace) -> None:
         args.web_rigctld = True
 
 
+def _bridge_gain_db(raw: str) -> float:
+    try:
+        gain = float(raw)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"{raw!r} is not a dB value") from None
+    if not isfinite(gain):
+        raise argparse.ArgumentTypeError(f"{raw!r} must be a finite dB value")
+    return gain
+
+
 def _ptt_hold_seconds(raw: str) -> float:
     """Parse ``--for``: a finite, positive number of seconds and nothing else.
 
@@ -1031,6 +1041,20 @@ def _add_audio_common_flags(sp: argparse.ArgumentParser) -> None:
         action="store_true",
         help="RX only (don't bridge TX from device to radio)",
     )
+    audio_bridge_p.add_argument(
+        "--input-gain-db",
+        type=_bridge_gain_db,
+        default=0.0,
+        metavar="DB",
+        help="Gain for device-to-radio audio in dB (default: 0.0)",
+    )
+    audio_bridge_p.add_argument(
+        "--output-gain-db",
+        type=_bridge_gain_db,
+        default=0.0,
+        metavar="DB",
+        help="Gain for radio-to-device audio in dB (default: 0.0)",
+    )
     audio_bridge_p.add_argument(
         "--list-devices",
         action="store_true",
@@ -1436,6 +1460,22 @@ def _add_audio_common_flags(sp: argparse.ArgumentParser) -> None:
         action="store_true",
         help="Bridge RX only (no TX from virtual device to radio)",
     )
+    web_p.add_argument(
+        "--bridge-input-gain-db",
+        dest="web_bridge_input_gain_db",
+        type=_bridge_gain_db,
+        default=0.0,
+        metavar="DB",
+        help="Bridge gain for device-to-radio audio in dB (default: 0.0)",
+    )
+    web_p.add_argument(
+        "--bridge-output-gain-db",
+        dest="web_bridge_output_gain_db",
+        type=_bridge_gain_db,
+        default=0.0,
+        metavar="DB",
+        help="Bridge gain for radio-to-device audio in dB (default: 0.0)",
+    )
     web_p.add_argument(
         "--bridge-label",
         dest="web_bridge_label",
@@ -3996,6 +4036,8 @@ async def _cmd_audio_bridge(radio: Radio, args: argparse.Namespace) -> int:
             radio,
             device_name=args.device,
             tx_device_name=getattr(args, "tx_device", None),
+            input_gain_db=getattr(args, "input_gain_db", 0.0),
+            output_gain_db=getattr(args, "output_gain_db", 0.0),
             tx_enabled=not args.rx_only,
             label=bridge_label,
             max_retries=getattr(args, "max_retries", 5),
@@ -4341,6 +4383,8 @@ async def _cmd_web(
             device_name=None,
             tx_device_name=getattr(args, "web_bridge_tx_device", None),
             tx_enabled=not getattr(args, "web_bridge_rx_only", False),
+            input_gain_db=getattr(args, "web_bridge_input_gain_db", 0.0),
+            output_gain_db=getattr(args, "web_bridge_output_gain_db", 0.0),
             label=getattr(args, "web_bridge_label", None),
             max_retries=getattr(args, "web_bridge_max_retries", 5),
             retry_base_delay=getattr(args, "web_bridge_retry_delay", 1.0),
@@ -4359,6 +4403,8 @@ async def _cmd_web(
                 device_name=device_name,
                 tx_device_name=tx_device_name,
                 tx_enabled=not rx_only,
+                input_gain_db=getattr(args, "web_bridge_input_gain_db", 0.0),
+                output_gain_db=getattr(args, "web_bridge_output_gain_db", 0.0),
                 label=bridge_label,
                 max_retries=getattr(args, "web_bridge_max_retries", 5),
                 retry_base_delay=getattr(args, "web_bridge_retry_delay", 1.0),

src/rigplane/web/api_contract.py

@@ -104,6 +104,12 @@ class ResponseFieldContract(TypedDict):
         "purpose": "start audio bridge",
         "auth": "none",
     },
+    {
+        "method": "PATCH",
+        "path": "/api/v1/bridge",
+        "purpose": "adjust running audio bridge gains",
+        "auth": "none",
+    },
     {
         "method": "DELETE",
         "path": "/api/v1/bridge",

src/rigplane/web/server.py

@@ -3388,6 +3388,9 @@ async def start_audio_bridge(
         label: str | None = None,
         max_retries: int = 5,
         retry_base_delay: float = 1.0,
+        *,
+        input_gain_db: float = 0.0,
+        output_gain_db: float = 0.0,
     ) -> None:
         """Start the audio bridge to a virtual audio device.
 
@@ -3396,6 +3399,8 @@ async def start_audio_bridge(
             tx_device_name: Separate device for TX (e.g. "RigPlane Virtual Cable Input").
                             Required for bidirectional audio to avoid feedback.
             tx_enabled: Whether to bridge TX (device → radio).
+            input_gain_db: Gain applied to device-to-radio PCM.
+            output_gain_db: Gain applied to radio-to-device PCM.
             label: Descriptive label for log messages. If ``None``, derived from radio model.
         """
         from rigplane.audio.bridge import AudioBridge, derive_bridge_label
@@ -3416,6 +3421,8 @@ async def start_audio_bridge(
             self._radio,  # type: ignore[arg-type]
             device_name=device_name,
             tx_device_name=tx_device_name,
+            input_gain_db=input_gain_db,
+            output_gain_db=output_gain_db,
             tx_enabled=tx_enabled,
             label=label,
             max_retries=max_retries,
@@ -3450,6 +3457,9 @@ def defer_audio_bridge_start(
         label: str | None = None,
         max_retries: int = 5,
         retry_base_delay: float = 1.0,
+        *,
+        input_gain_db: float = 0.0,
+        output_gain_db: float = 0.0,
     ) -> None:
         """Arm ONE deferred auto-start of the audio bridge (MOR-3078).
 
@@ -3466,6 +3476,8 @@ def defer_audio_bridge_start(
             "device_name": device_name,
             "tx_device_name": tx_device_name,
             "tx_enabled": tx_enabled,
+            "input_gain_db": input_gain_db,
+            "output_gain_db": output_gain_db,
             "label": label,
             "max_retries": max_retries,
             "retry_base_delay": retry_base_delay,
@@ -3477,6 +3489,8 @@ async def _run_deferred_audio_bridge_start(
         device_name: str | None,
         tx_device_name: str | None,
         tx_enabled: bool,
+        input_gain_db: float,
+        output_gain_db: float,
         label: str | None,
         max_retries: int,
         retry_base_delay: float,
@@ -3498,6 +3512,8 @@ async def _run_deferred_audio_bridge_start(
                 device_name=device_name,
                 tx_device_name=tx_device_name,
                 tx_enabled=tx_enabled,
+                input_gain_db=input_gain_db,
+                output_gain_db=output_gain_db,
                 label=label,
                 max_retries=max_retries,
                 retry_base_delay=retry_base_delay,
@@ -6912,13 +6928,20 @@ async def _handle_bridge(
         self,
         method: str,
         writer: asyncio.StreamWriter,
+        headers: dict[str, str] | None = None,
+        reader: asyncio.StreamReader | None = None,
     ) -> None:
-        """Handle /api/v1/bridge — GET status, POST start, DELETE stop."""
+        """Handle bridge status, lifecycle, and dynamic gain updates."""
         if method == "GET":
             stats = self.audio_bridge_stats
+            bridge = self._audio_bridge
             body = json.dumps(
                 {
                     "running": stats is not None and stats.get("running", False),
+                    "inputGainDb": bridge.input_gain_db if bridge is not None else 0.0,
+                    "outputGainDb": (
+                        bridge.output_gain_db if bridge is not None else 0.0
+                    ),
                     **(stats or {}),
                 },
                 separators=(",", ":"),
@@ -6953,6 +6976,65 @@ async def _handle_bridge(
                     body,
                     {"Content-Type": "application/json"},
                 )
+        elif method == "PATCH":
+            bridge = self._audio_bridge
+            if bridge is None or not bridge.running:
+                await _send_diag_error(
+                    writer,
+                    409,
+                    "bridge_not_running",
+                    "audio bridge must be running to adjust gain",
+                )
+                return
+            payload = await self._read_json_body(writer, headers, reader)
+            if payload is None:
+                return
+            allowed = {"inputGainDb", "outputGainDb"}
+            if not payload or set(payload) - allowed:
+                await _send_diag_error(
+                    writer,
+                    400,
+                    "invalid_bridge_gain",
+                    "inputGainDb or outputGainDb is required",
+                )
+                return
+
+            gains: dict[str, float] = {}
+            for key in allowed & set(payload):
+                value = payload[key]
+                if (
+                    isinstance(value, bool)
+                    or not isinstance(value, (int, float))
+                    or abs(value) > sys.float_info.max
+                    or not math.isfinite(value)
+                ):
+                    await _send_diag_error(
+                        writer,
+                        400,
+                        "invalid_bridge_gain",
+                        f"{key} must be a finite number",
+                    )
+                    return
+                gains[key] = float(value)
+
+            bridge.set_gains(
+                input_gain_db=gains.get("inputGainDb"),
+                output_gain_db=gains.get("outputGainDb"),
+            )
+            body = json.dumps(
+                {
+                    "inputGainDb": bridge.input_gain_db,
+                    "outputGainDb": bridge.output_gain_db,
+                },
+                separators=(",", ":"),
+            ).encode()
+            await _send_response(
+                writer,
+                200,
+                "OK",
+                body,
+                {"Content-Type": "application/json"},
+            )
         elif method == "DELETE":
             await self.stop_audio_bridge()
             body = json.dumps({"status": "stopped"}, separators=(",", ":")).encode()

src/rigplane/web/web_routing.py

@@ -234,10 +234,10 @@ async def _dispatch_http_routes(
         return
 
     if path == "/api/v1/bridge":
-        if method not in ("GET", "HEAD", "POST", "DELETE"):
+        if method not in ("GET", "HEAD", "POST", "PATCH", "DELETE"):
             await _send_response(writer, 405, "Method Not Allowed", b"", {})
             return
-        await server._handle_bridge(method, writer)
+        await server._handle_bridge(method, writer, headers, reader)
         return
     if path in (
         "/api/v1/radio/disconnect",