fix(repeater): the receive window could not contain a frame (#1297)

CLAUDE.md

@@ -239,6 +239,7 @@ Each requirement below is done when the linked test passes. Add new links as tes
 | `[radio.rig_b]` cannot alias the rigctld the main rig already uses — the daemon refuses to start **when the repeater is enabled at startup** (otherwise it warns and builds no repeater). Two controllers over one transmitter key and release each other, and #1263's refusal rule reaches only *within* one `SharedPtt`. Not exotic: both `RigConfig::default()` and `RadioConfig::default()` carry `127.0.0.1:4532`, so an empty `[radio.rig_b]` header IS the collision — and the shared endpoint is rigctld itself, so `cat_backend = "rigctld"` alone collides even with a non-rigctld `ptt_backend`. Scoped to string equality: it catches the shipped defaults, not `localhost` vs `127.0.0.1` | `cargo test -p openpulse-daemon --no-default-features --lib repeater_rig_b_tests` |
 | A **cap-flushed** burst is not evidence about the rate ladder (#1255) — the cap exceeds the longest candidate frame, so hitting it means the carrier was still up and the slab is not one transmission. A failed decode of one must not key an ACK or move `recommended_level`. The decode itself still runs: when the squelch sits below the band floor EVERY burst is a cap flush (#1254's regime), so skipping it would make the daemon deaf on a hot band — pinned by a control that decodes a frame at the head of a capped slab. Runs ~70 s, dominated by one `ota_decode_burst` over the candidate rungs | `cargo test -p openpulse-modem --no-default-features --test cap_flush_is_not_ladder_evidence` |
 | A repeater that is not running is not reported as running (#1298) — enabling with nothing to run FAILS with a reason instead of emitting `RepeaterChanged { enabled: true }`, and a thread that exited is reaped so the next command sees the truth rather than "already enabled" forever. The thread OWNS the `CrossBandRepeater`, so its exit means the repeater is gone | `cargo test -p openpulse-daemon --no-default-features --lib command_apply_tests` |
+| The cross-band repeater can receive a frame that arrives across SEVERAL reads (#1297) — it holds one capture stream and accumulates under DCD gating instead of opening a stream, reading once and dropping it, which on a callback backend gave it a window of one poll interval against a seconds-long frame. `LoopbackBackend` hides this entirely (its read drains the whole buffer), so the gate delivers the frame one chunk per read via `push_frame`; sabotage-verified against the old `receive()` path | `cargo test -p openpulse-repeater --no-default-features --test repeater_integration a_frame_split_across_several_reads_is_still_relayed` |
 | `openpulse-kiss`'s `SharedPtt` has a **watchdog thread**, so its deadline is enforced — the crate built one and called `spawn_watchdog` nowhere, leaving `force_release_if_expired` with no caller in the crate. The guard covers an early return and an unwind; it cannot reach a transmit that BLOCKS, which is the case the watchdog exists for. Driven through the real constructor, since the defect was the wiring | `cargo test -p openpulse-kiss --no-default-features --test ptt_keys_every_transmit` |
 | `openpulse-mesh` has no route to a sound card — it beacons and relays automatically with no PTT controller, no carrier sense and no station-ID timer, and its beacon carries no callsign field, so the capability was REMOVED rather than guarded (a fourth hand-rolled keying path on a crate with no §97.221 mapping, no control point and no on-air record). Each check is validated against a planted input | `cargo test -p openpulse-mesh --no-default-features --test no_real_audio` |
 | A CONACK cannot select a signing mode the CONREQ never offered (F-1147-05 — v1 checked local policy only) | `cargo test -p openpulse-core --no-default-features --test handshake_integration conack_rejected_when_mode_not_offered` |

crates/openpulse-modem/src/capture_ticker.rs

@@ -0,0 +1,143 @@
+//! One held-open capture stream, read a tick at a time and folded into the engine's burst
+//! accumulator (#1297).
+//!
+//! **Why this exists rather than `ModemEngine::receive`.** `receive` opens an input stream, reads
+//! once, and drops the stream — so on a callback backend each call sees a fresh, nearly-empty buffer
+//! covering one poll interval. A frame is seconds long; a poll is tens of milliseconds. Any caller
+//! that listens *continuously* therefore cannot receive at all on real audio, however long it runs,
+//! and `capture_burst`'s own comment records the reason: a fresh cpal stream needs tens of ms to
+//! start delivering, so reopening per tick never warms up.
+//!
+//! `LoopbackBackend` hides this completely — its `read` drains the whole buffer, so the buffer IS
+//! the frame and `receive` works. That is why the defect survived a green suite.
+//!
+//! The daemon's `rx_ticker` already does the right thing by hand. This is that pattern, owned once:
+//! open lazily, read, accumulate, and on a fault drop the stream so the next tick reopens it —
+//! reporting the first failure at WARN, because a station that cannot hear is off the air and
+//! `receive`'s error path made that indistinguishable from a quiet band.
+//!
+//! **Twin copies that should adopt this.** `server.rs`'s `rx_ticker` (`:886-1030`) is the original,
+//! open-coded, and still carries its own `block_in_place` wrapper, discovery tee and logging — it is
+//! left alone here because refactoring a working receive path inside a PR that fixes a broken one
+//! trades risk for tidiness. `openpulse-ardop` (`bridge.rs:430,437,557,564`) and `openpulse-kiss`
+//! (`bridge.rs:237,256`) call `receive` in free-running loops and have the SAME structural defect
+//! this fixes — the assumption that their TCP-driven shape makes a per-call window correct is false;
+//! both poll continuously. Tracked as a follow-up rather than fixed blind.
+
+use openpulse_core::audio::AudioInputStream;
+use openpulse_core::error::ModemError;
+
+use crate::engine::ModemEngine;
+use crate::pipeline::AudioSamples;
+
+/// What one tick produced.
+pub struct Tick {
+    /// A burst, when the accumulator flushed one this tick.
+    pub burst: Option<AudioSamples>,
+    /// The raw samples this tick read, before the RX front end — for callers that tee the audio
+    /// (the daemon's JS8 discovery dwell). Empty when the read produced nothing.
+    pub raw: Vec<f32>,
+}
+
+/// A capture stream held across ticks, feeding [`ModemEngine::accumulate_capture`].
+pub struct CaptureTicker {
+    stream: Option<Box<dyn AudioInputStream>>,
+    device: Option<String>,
+    /// Whether the last read or open failed, so the recurring case logs at DEBUG and the first
+    /// failure and the recovery each log at WARN.
+    failed: bool,
+}
+
+impl CaptureTicker {
+    /// Capture from `device`, or the engine's default when `None`.
+    pub fn new(device: Option<String>) -> Self {
+        Self {
+            stream: None,
+            device,
+            failed: false,
+        }
+    }
+
+    /// Whether the stream is currently faulted; the next tick will try to reopen.
+    pub fn is_faulted(&self) -> bool {
+        self.failed
+    }
+
+    /// Read one tick and fold it into `engine`'s burst accumulator.
+    ///
+    /// Never returns the open/read error: a capture fault is reported and retried, not propagated,
+    /// because a caller that treats it as fatal stops listening for good. Decode-side errors are the
+    /// caller's business and reach it through the returned burst.
+    pub fn tick(&mut self, engine: &mut ModemEngine, mode: &str) -> Tick {
+        if self.stream.is_none() {
+            match engine.open_capture_stream(self.device.as_deref()) {
+                Ok(s) => {
+                    if self.failed {
+                        self.failed = false;
+                        tracing::warn!("audio capture recovered");
+                    }
+                    self.stream = Some(s);
+                }
+                Err(e) => {
+                    self.note_fault(&e, "cannot open the capture device");
+                    return Tick {
+                        burst: None,
+                        raw: Vec::new(),
+                    };
+                }
+            }
+        }
+
+        let read = match self.stream.as_mut() {
+            Some(s) => s.read(),
+            None => {
+                return Tick {
+                    burst: None,
+                    raw: Vec::new(),
+                }
+            }
+        };
+
+        match read {
+            Ok(samples) => {
+                if self.failed {
+                    self.failed = false;
+                    tracing::warn!("audio capture recovered");
+                }
+                let raw = samples.clone();
+                let burst = engine
+                    .accumulate_capture(Some(mode), samples)
+                    .unwrap_or_else(|e| {
+                        // The front end failed on this block; not a capture fault, and not worth
+                        // dropping the stream for.
+                        tracing::debug!(error = %e, "capture accumulate failed for one block");
+                        None
+                    });
+                Tick { burst, raw }
+            }
+            Err(e) => {
+                // Drop the stream so the next tick reopens it.
+                self.stream = None;
+                self.note_fault(
+                    &ModemError::Audio(e.to_string()),
+                    "audio capture read failed",
+                );
+                Tick {
+                    burst: None,
+                    raw: Vec::new(),
+                }
+            }
+        }
+    }
+
+    fn note_fault(&mut self, e: &ModemError, what: &str) {
+        if self.failed {
+            tracing::debug!(error = %e, "{what}; still failing, will retry");
+        } else {
+            self.failed = true;
+            // WARN, not DEBUG: this is the state in which a station hears nothing at all, and it
+            // must not look like a quiet band.
+            tracing::warn!(error = %e, "{what}; retrying on the next tick");
+        }
+    }
+}

crates/openpulse-modem/src/lib.rs

@@ -5,6 +5,7 @@
 
 pub mod benchmark;
 pub mod capture_replay;
+pub mod capture_ticker;
 pub mod channel_sim;
 pub mod diagnostics;
 pub mod engine;

crates/openpulse-repeater/src/lib.rs

@@ -6,6 +6,7 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use openpulse_core::station_id::StationIdTimer;
+use openpulse_modem::capture_ticker::CaptureTicker;
 use openpulse_modem::ModemEngine;
 use openpulse_radio::{PttController, PttKeyGuard, SharedPtt, DEFAULT_PTT_MAX};
 use thiserror::Error;
@@ -85,25 +86,42 @@ impl CrossBandRepeater {
     ///
     /// Returns the number of bytes relayed, or `None` if no frame was available.
     /// FEC is not applied on the relay path (raw mode).
-    pub fn relay_one_frame(&mut self) -> Result<Option<usize>, RepeaterError> {
+    pub fn relay_one_frame(
+        &mut self,
+        rx: &mut CaptureTicker,
+    ) -> Result<Option<usize>, RepeaterError> {
         let now_ms = self.start.elapsed().as_millis() as u64;
-        self.relay_one_frame_at(now_ms)
+        self.relay_one_frame_at(rx, now_ms)
     }
 
     /// [`relay_one_frame`] with an explicit monotonic clock (for deterministic ID-timing tests).
-    pub fn relay_one_frame_at(&mut self, now_ms: u64) -> Result<Option<usize>, RepeaterError> {
+    pub fn relay_one_frame_at(
+        &mut self,
+        rx: &mut CaptureTicker,
+        now_ms: u64,
+    ) -> Result<Option<usize>, RepeaterError> {
         if !self.config.enabled {
             return Ok(None);
         }
 
-        let bytes = match self.engine_rx.receive(&self.config.mode.clone(), None) {
+        // Hold ONE capture stream across attempts and accumulate under DCD gating, the way the
+        // daemon's rx ticker does (#1297). A capture fault is reported and retried inside the
+        // ticker rather than surfacing here: treating it as fatal would stop the repeater
+        // listening for good, and treating it as silence — which the previous `Err => Ok(None)`
+        // arm did — made an unopenable RX device indistinguishable from a quiet band, at DEBUG.
+        let Some(burst) = rx.tick(&mut self.engine_rx, &self.config.mode).burst else {
+            return Ok(None);
+        };
+
+        let bytes = match self
+            .engine_rx
+            .decode_burst(&self.config.mode.clone(), &burst)
+        {
             Ok(b) => b,
             Err(e) => {
-                // A capture that does not demodulate is indistinguishable from an idle channel — it
-                // is the ordinary quiet case, not a fault. Propagating it ended the whole session on
-                // the first window with no frame in it (#1297); only PTT and transmit faults stop a
-                // repeater.
-                tracing::debug!(error = %e, "cross-band relay: no frame in this capture window");
+                // A burst that does not decode is the ordinary case on a live band: noise that
+                // opened the squelch, or a frame this repeater's mode cannot read. Not a fault.
+                tracing::debug!(error = %e, "cross-band relay: burst did not decode");
                 return Ok(None);
             }
         };
@@ -205,12 +223,17 @@ impl CrossBandRepeater {
         if !self.config.enabled {
             return Ok(0);
         }
+        // The capture stream is owned HERE, not on the struct: `Box<dyn AudioInputStream>` is not
+        // `Send` (a cpal `Stream` is not, on most hosts) and the daemon moves the repeater into a
+        // thread, so a stream field would make `CrossBandRepeater` unspawnable. The daemon's own rx
+        // ticker keeps its stream as a loop local for the same reason.
+        let mut rx = CaptureTicker::new(None);
         let mut count = 0u64;
         let result = loop {
             if stop.load(Ordering::Relaxed) {
                 break Ok(count);
             }
-            match self.relay_one_frame() {
+            match self.relay_one_frame(&mut rx) {
                 Ok(Some(_)) => count += 1,
                 // Since #1297 an idle window is `Ok(None)` rather than a session-ending error, so
                 // this arm is now reached continuously instead of never. Without a pause the loop
@@ -270,6 +293,22 @@ mod full_duplex_silence_tests {
         e
     }
 
+    /// Tick until a frame is relayed, the way the daemon's loop does.
+    ///
+    /// Since #1297 one `relay_one_frame_at` is one capture TICK, not one receive attempt: the burst
+    /// accumulator flushes when the carrier drops, which on `LoopbackBackend` is the first empty
+    /// read after the frame. So relaying takes at least two calls.
+    fn relay_until(rp: &mut CrossBandRepeater, now_ms: u64) -> usize {
+        let mut rx = CaptureTicker::new(None);
+        for _ in 0..16 {
+            match rp.relay_one_frame_at(&mut rx, now_ms).expect("relay") {
+                Some(n) => return n,
+                None => continue,
+            }
+        }
+        panic!("no frame relayed within 16 ticks");
+    }
+
     /// The half of "the deadline measures silence" that no integration test can reach.
     ///
     /// `acquire_key`'s re-key branch fires only after the watchdog has taken a held key, and the
@@ -306,7 +345,7 @@ mod full_duplex_silence_tests {
         };
 
         feed();
-        rp.relay_one_frame_at(0).expect("relay").expect("Some");
+        relay_until(&mut rp, 0);
         assert_eq!(
             *spy.edges.lock().expect("lock"),
             vec!["assert"],
@@ -323,7 +362,7 @@ mod full_duplex_silence_tests {
         );
 
         feed();
-        rp.relay_one_frame_at(1_000).expect("relay").expect("Some");
+        relay_until(&mut rp, 1_000);
         assert_eq!(
             *spy.edges.lock().expect("lock"),
             vec!["assert", "release", "assert"],
@@ -358,9 +397,7 @@ mod full_duplex_silence_tests {
             src.register_plugin(Box::new(BpskPlugin::new()))
                 .expect("register src");
             src.transmit(b"fd frame", "BPSK250", None).expect("tx");
-            rp.relay_one_frame_at(i * 100)
-                .expect("relay")
-                .expect("Some");
+            relay_until(&mut rp, i * 100);
             std::thread::sleep(Duration::from_millis(100));
         }
 

crates/openpulse-repeater/tests/half_duplex_ptt.rs

@@ -160,9 +160,20 @@ fn a_failed_relay_transmit_does_not_leave_the_transmitter_keyed() {
     src.transmit(b"relay frame", "BPSK250", None).expect("tx");
 
     let mut rp = CrossBandRepeater::new(Box::new(spy.clone()), engine_rx, engine_tx, config);
-    let err = rp
-        .relay_one_frame()
-        .expect_err("the tx engine has no plugin, so the relay must fail");
+    // Since #1297 one call is one capture TICK: the burst flushes on the first empty read after the
+    // frame, so the transmit — and its failure — happen on a later tick than the one that read it.
+    let mut rx = openpulse_modem::capture_ticker::CaptureTicker::new(None);
+    let mut err = None;
+    for _ in 0..16 {
+        match rp.relay_one_frame(&mut rx) {
+            Ok(_) => continue,
+            Err(e) => {
+                err = Some(e);
+                break;
+            }
+        }
+    }
+    let err = err.expect("the tx engine has no plugin, so the relay must fail within 16 ticks");
 
     assert!(
         !spy.is_asserted(),