clawft_kernel/
thin_edge.rs

1//! Desktop thin edge over localhost — Phase 1 scaffold (ADR-068 / WEFT-628).
2//!
3//! The edge is a **dumb streaming transducer**: capture + playback + a minimal
4//! reflex set + a control/media lane pair. Cognition (STT/TTS/VAD/endpoint/
5//! floor/classify) lives daemon-side. This module defines the **localhost wire
6//! vocabulary** and an **in-process loopback session** that proves:
7//!
8//! 1. The edge streams [`EdgeState`] (incl. ERL confidence) up the control lane.
9//! 2. The daemon [`DuplexChannel`] folds ERL into `TurnClaim`s.
10//! 3. [`compute_urgency`](crate::floor::compute_urgency) / [`floor_verdict_from_erl`]
11//!    weight low-ERL claims down so self-echo cannot win the floor.
12//!
13//! No cpal, no network, no TalkForest: pure protocol + state-machine wiring so
14//! later daemon RPC / UDS transport (plan Phase 1) plugs in without reshaping
15//! types. When the daemon is up the edge holds **no** standalone in-memory
16//! forest — that invariant is structural here (this session only owns
17//! [`DuplexChannel`] + edge reflex shadow flags).
18
19use crate::duplex::{
20    DuplexChannel, DuplexImpulse, DuplexState, EdgeCommand, FloorVerdict, PayloadKind,
21    StreamObservation,
22};
23use crate::floor::{
24    ContentReadiness, ErlBargeConfig, ErlBargeGate, UrgencySignals, compute_urgency,
25    floor_verdict_from_erl, floor_verdict_from_erl_config,
26};
27use crate::impulse::ImpulseType;
28
29// ---------------------------------------------------------------------------
30// Wire vocabulary (ADR-068 D3) — payload-typed media + control lanes
31// ---------------------------------------------------------------------------
32
33/// Media-lane frame the edge streams up (mic / keystrokes) or the daemon
34/// streams down (TTS / rendered text). Typed by negotiated capability, not by
35/// a baked profile (ADR-068 D2).
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub enum MediaFrame {
38    /// 16 kHz PCM (or host sample-rate) bytes.
39    Pcm(Vec<u8>),
40    /// UTF-8 text deltas (terminal / GUI chat / mid-profile STT offload).
41    Text(String),
42}
43
44/// Control-lane state the edge reports up every tick / on change (ADR-068 D3).
45#[derive(Debug, Clone, Copy, PartialEq)]
46pub struct EdgeState {
47    /// Speaker path currently playing daemon media.
48    pub playback_active: bool,
49    /// Latest AEC echo-return-loss confidence in `[0, 1]`, when measured.
50    pub erl_confidence: Option<f32>,
51    /// Silence-squelch engaged (bandwidth only; no semantic authority).
52    pub squelch_active: bool,
53    /// Heartbeat sequence so the daemon can detect link loss.
54    pub heartbeat: u64,
55}
56
57impl EdgeState {
58    /// Fresh edge: no playback, no ERL, no squelch, heartbeat 0.
59    pub const fn new() -> Self {
60        Self {
61            playback_active: false,
62            erl_confidence: None,
63            squelch_active: false,
64            heartbeat: 0,
65        }
66    }
67}
68
69impl Default for EdgeState {
70    fn default() -> Self {
71        Self::new()
72    }
73}
74
75/// Control-lane message (either direction).
76#[derive(Debug, Clone, PartialEq)]
77pub enum ControlMessage {
78    /// Edge → daemon: current reflex/AEC shadow.
79    EdgeUp(EdgeState),
80    /// Daemon → edge: concrete command (floor verdicts never leave the daemon).
81    Cmd(EdgeCommand),
82}
83
84// ---------------------------------------------------------------------------
85// Thin edge shadow (edge-side of the localhost session)
86// ---------------------------------------------------------------------------
87
88/// Minimal edge-side shadow of the three ADR-068 D5 reflexes, mirrored here so
89/// the kernel crate can unit-test the loopback without depending on
90/// `clawft-channels`. The channels `EdgeReflex` is the production type; this
91/// is a byte-compatible state shadow for the scaffold.
92#[derive(Debug, Clone, Copy, PartialEq, Eq)]
93pub enum EdgeReflexShadow {
94    /// Streaming mic up, speaker permitted.
95    Streaming,
96    /// Speaker silenced by daemon command.
97    Muted,
98    /// Link lost — no TX, no speaker.
99    Degraded,
100}
101
102/// Edge half of the localhost duplex session (WEFT-628).
103///
104/// Holds only reflex state + the last ERL; no forest, no STT, no floor
105/// decisions. Daemon commands arrive as [`EdgeCommand`]; ERL is observed and
106/// reported, never judged.
107#[derive(Debug, Clone)]
108pub struct ThinEdge {
109    reflex: EdgeReflexShadow,
110    state: EdgeState,
111    tx: bool,
112    speaker: bool,
113}
114
115impl ThinEdge {
116    /// Fresh thin edge: streaming, speaker on, no ERL.
117    pub fn new() -> Self {
118        Self {
119            reflex: EdgeReflexShadow::Streaming,
120            state: EdgeState::new(),
121            tx: true,
122            speaker: true,
123        }
124    }
125
126    /// Current reflex shadow.
127    pub fn reflex(&self) -> EdgeReflexShadow {
128        self.reflex
129    }
130
131    /// Latest control-lane snapshot (clone of the live state).
132    pub fn edge_state(&self) -> EdgeState {
133        let mut s = self.state;
134        s.playback_active = self.speaker && self.reflex != EdgeReflexShadow::Degraded;
135        s
136    }
137
138    /// Whether the edge is transmitting mic frames up.
139    pub fn is_transmitting(&self) -> bool {
140        self.tx && self.reflex != EdgeReflexShadow::Degraded
141    }
142
143    /// Relay AEC ERL confidence (ADR-068 D5 #1 is the AEC itself; this is the
144    /// control-lane relay only).
145    pub fn observe_erl(&mut self, erl: f32) {
146        self.state.erl_confidence = Some(erl);
147    }
148
149    /// Execute a daemon command (instant mute/silence/resume/cancel).
150    pub fn on_command(&mut self, cmd: &EdgeCommand) {
151        if self.reflex == EdgeReflexShadow::Degraded {
152            return;
153        }
154        match cmd {
155            EdgeCommand::Mute | EdgeCommand::Silence | EdgeCommand::Cancel => {
156                self.reflex = EdgeReflexShadow::Muted;
157                self.speaker = false;
158            }
159            EdgeCommand::Resume | EdgeCommand::Play(_) => {
160                self.reflex = EdgeReflexShadow::Streaming;
161                self.speaker = true;
162            }
163            EdgeCommand::Config => {}
164        }
165    }
166
167    /// Link-loss safety (the one self-firing rule).
168    pub fn on_link_loss(&mut self) {
169        self.reflex = EdgeReflexShadow::Degraded;
170        self.tx = false;
171        self.speaker = false;
172    }
173
174    /// Re-handshake after link restore: TX resumes, speaker stays silent until
175    /// the daemon plays (never speak unprompted).
176    pub fn on_link_restored(&mut self) {
177        self.reflex = EdgeReflexShadow::Streaming;
178        self.tx = true;
179        self.speaker = false;
180        self.state.erl_confidence = None;
181        self.state.heartbeat = self.state.heartbeat.saturating_add(1);
182    }
183
184    /// Tick the heartbeat (edge → daemon keep-alive).
185    pub fn beat(&mut self) {
186        self.state.heartbeat = self.state.heartbeat.saturating_add(1);
187    }
188}
189
190impl Default for ThinEdge {
191    fn default() -> Self {
192        Self::new()
193    }
194}
195
196// ---------------------------------------------------------------------------
197// Localhost duplex session (daemon + edge, in-process)
198// ---------------------------------------------------------------------------
199
200/// Outcome of one observation step on the daemon half.
201#[derive(Debug, Clone, PartialEq)]
202pub struct ObserveOutcome {
203    /// Impulse lifted onto the ECC queue (if any).
204    pub impulse: Option<DuplexImpulse>,
205    /// Commands delivered to the edge this step.
206    pub commands: Vec<EdgeCommand>,
207    /// Daemon duplex state after the observation.
208    pub duplex_state: DuplexState,
209}
210
211/// In-process localhost thin-edge session (ADR-068 Phase 1 scaffold).
212///
213/// The edge half is a [`ThinEdge`]; the daemon half is a [`DuplexChannel`].
214/// There is **no** private TalkForest — floor cognition rides the shared
215/// daemon machine only. Transport is a pair of `Vec`s simulating the two
216/// lanes; a later UDS/WebSocket carrier swaps the vectors for sockets.
217///
218/// The session holds an [`ErlBargeGate`] (WEFT-615) so repeated ERL samples
219/// during provisional Overlap use floor + hysteresis rather than a binary
220/// barge switch.
221#[derive(Debug)]
222pub struct LocalhostDuplexSession {
223    /// Edge-side reflex + ERL relay.
224    pub edge: ThinEdge,
225    /// Daemon-side floor actuator.
226    pub duplex: DuplexChannel,
227    /// Stateful ERL barge gate (floor + hysteresis knobs).
228    pub erl_gate: ErlBargeGate,
229    /// Media frames the edge has streamed up (for inspection / later STT).
230    pub media_up: Vec<MediaFrame>,
231    /// Media frames the daemon has streamed down (TTS / text).
232    pub media_down: Vec<MediaFrame>,
233}
234
235impl LocalhostDuplexSession {
236    /// Open a localhost session for the given payload axis (default ERL knobs).
237    pub fn open(payload: PayloadKind) -> Self {
238        Self::open_with_erl(payload, ErlBargeConfig::default())
239    }
240
241    /// Open with explicit ERL barge floor + hysteresis (WEFT-615 knobs).
242    pub fn open_with_erl(payload: PayloadKind, erl: ErlBargeConfig) -> Self {
243        Self {
244            edge: ThinEdge::new(),
245            duplex: DuplexChannel::new(payload),
246            erl_gate: ErlBargeGate::new(erl),
247            media_up: Vec::new(),
248            media_down: Vec::new(),
249        }
250    }
251
252    /// Edge pushes a media frame up (thin client of the daemon stream).
253    pub fn edge_push_media(&mut self, frame: MediaFrame) {
254        if self.edge.is_transmitting() {
255            self.media_up.push(frame);
256        }
257    }
258
259    /// Edge reports its control state (ERL + heartbeat) into the daemon.
260    ///
261    /// Folds `erl_confidence` into the duplex machine as
262    /// [`StreamObservation::Erl`] so subsequent `TurnClaim`s carry it.
263    pub fn edge_report_state(&mut self) -> ObserveOutcome {
264        self.edge.beat();
265        let state = self.edge.edge_state();
266        if let Some(erl) = state.erl_confidence {
267            let (_imp, cmds) = self.duplex.observe(StreamObservation::Erl(erl));
268            debug_assert!(cmds.is_empty(), "Erl observation is non-transitional");
269        }
270        ObserveOutcome {
271            impulse: None,
272            commands: Vec::new(),
273            duplex_state: self.duplex.state(),
274        }
275    }
276
277    /// Daemon-local stream observation (VAD/EOU/drain/heartbeat), delivering
278    /// any resulting commands to the edge.
279    pub fn daemon_observe(&mut self, obs: StreamObservation) -> ObserveOutcome {
280        let (impulse, commands) = self.duplex.observe(obs);
281        for cmd in &commands {
282            self.edge.on_command(cmd);
283        }
284        ObserveOutcome {
285            impulse,
286            commands,
287            duplex_state: self.duplex.state(),
288        }
289    }
290
291    /// Apply an ECC floor verdict and deliver edge commands.
292    pub fn apply_verdict(&mut self, verdict: FloorVerdict) -> Vec<EdgeCommand> {
293        let commands = self.duplex.apply(verdict);
294        for cmd in &commands {
295            self.edge.on_command(cmd);
296        }
297        commands
298    }
299
300    /// Score a barge claim for the floor using the claim's ERL (WEFT-628).
301    ///
302    /// Builds a contender urgency signal with the ERL scale applied so a
303    /// low-ERL self-echo claim cannot outscore an agent holding the floor.
304    pub fn score_barge_claim(impulse: &DuplexImpulse) -> f32 {
305        if impulse.kind != ImpulseType::TurnClaim {
306            return 0.0;
307        }
308        let signals = UrgencySignals {
309            semantic_relevance: 1.0, // onset is maximally relevant to the floor
310            emotional_arousal: 0.5,
311            wait_time: 0.0,
312            crowd_density: 0.25,
313            content_readiness: ContentReadiness::NotStarted,
314            hard_interrupt: false,
315            erl_confidence: impulse.erl,
316        };
317        compute_urgency(&signals)
318    }
319
320    /// JSON payload for lifting a duplex `TurnClaim` onto the ECC
321    /// [`ImpulseQueue`](crate::impulse::ImpulseQueue) (WEFT-615).
322    ///
323    /// Always carries `erl_confidence` as a number so the Talk-Mode loop's
324    /// acoustic gate fires: missing measurement becomes `0.0` (deny / safe).
325    pub fn duplex_turn_claim_payload(impulse: &DuplexImpulse) -> serde_json::Value {
326        serde_json::json!({
327            "erl_confidence": impulse.erl.unwrap_or(0.0),
328        })
329    }
330
331    /// Decide a provisional Overlap using ERL only (scaffold for TalkModeLoop
332    /// wiring). High ERL → grant user + mute; low ERL → deny barge.
333    ///
334    /// Uses default knobs (stateless one-shot). Prefer
335    /// [`resolve_overlap_with_gate`](Self::resolve_overlap_with_gate) when
336    /// sampling ERL repeatedly during Overlap (hysteresis).
337    pub fn resolve_overlap_from_erl(&mut self, erl: Option<f32>) -> (FloorVerdict, Vec<EdgeCommand>) {
338        let verdict = floor_verdict_from_erl(erl);
339        if matches!(verdict, FloorVerdict::DenyBarge) {
340            self.erl_gate.reset();
341        }
342        let cmds = self.apply_verdict(verdict);
343        (verdict, cmds)
344    }
345
346    /// Resolve Overlap through the session's stateful [`ErlBargeGate`]
347    /// (WEFT-615 floor + hysteresis). Feed successive ERL samples during
348    /// provisional Overlap; boundary flutter will not thrash Mute/Resume.
349    pub fn resolve_overlap_with_gate(
350        &mut self,
351        erl: Option<f32>,
352    ) -> (FloorVerdict, Vec<EdgeCommand>) {
353        let admitted = match erl {
354            Some(v) => self.erl_gate.observe(v),
355            None => {
356                self.erl_gate.reset();
357                false
358            }
359        };
360        let verdict = if admitted {
361            FloorVerdict::GrantUser
362        } else {
363            FloorVerdict::DenyBarge
364        };
365        let cmds = self.apply_verdict(verdict);
366        (verdict, cmds)
367    }
368
369    /// Stateless resolve with explicit knobs (no session gate state).
370    pub fn resolve_overlap_with_config(
371        &mut self,
372        erl: Option<f32>,
373        config: &ErlBargeConfig,
374    ) -> (FloorVerdict, Vec<EdgeCommand>) {
375        let verdict = floor_verdict_from_erl_config(erl, config);
376        let cmds = self.apply_verdict(verdict);
377        (verdict, cmds)
378    }
379}
380
381// ---------------------------------------------------------------------------
382// Tests
383// ---------------------------------------------------------------------------
384
385#[cfg(test)]
386mod tests {
387    use super::*;
388    use crate::floor::erl_admits_barge;
389
390    #[test]
391    fn thin_edge_has_no_forest_only_reflex() {
392        let e = ThinEdge::new();
393        assert_eq!(e.reflex(), EdgeReflexShadow::Streaming);
394        assert!(e.is_transmitting());
395        // Structural: ThinEdge fields are reflex + EdgeState only — no graph.
396        let _ = e.edge_state();
397    }
398
399    #[test]
400    fn erl_relays_into_duplex_turn_claim() {
401        let mut s = LocalhostDuplexSession::open(PayloadKind::Audio);
402        s.edge.observe_erl(0.12);
403        s.edge_report_state();
404        assert_eq!(s.duplex.last_erl(), Some(0.12));
405
406        // Idle → Listening on onset; claim carries the low ERL.
407        let out = s.daemon_observe(StreamObservation::VadOnset);
408        let claim = out.impulse.expect("TurnClaim");
409        assert_eq!(claim.kind, ImpulseType::TurnClaim);
410        assert_eq!(claim.erl, Some(0.12));
411    }
412
413    #[test]
414    fn self_cancel_low_erl_denies_barge_no_mute() {
415        let mut s = LocalhostDuplexSession::open(PayloadKind::Audio);
416        // Agent speaking.
417        let _ = s.daemon_observe(StreamObservation::VadOnset); // → Listening
418        let _ = s.daemon_observe(StreamObservation::SmartTurnEou); // → Thinking
419        let _ = s.apply_verdict(FloorVerdict::GrantAgent); // → Speaking
420        assert_eq!(s.duplex.state(), DuplexState::Speaking);
421
422        // Self-echo onset with low ERL → provisional Overlap.
423        s.edge.observe_erl(0.08);
424        s.edge_report_state();
425        let out = s.daemon_observe(StreamObservation::VadOnset);
426        assert_eq!(out.duplex_state, DuplexState::Overlap);
427        let claim = out.impulse.expect("TurnClaim");
428        assert!(!erl_admits_barge(claim.erl.unwrap_or(0.0)));
429
430        // ERL-scaled urgency is tiny — cannot win a contested floor.
431        let score = LocalhostDuplexSession::score_barge_claim(&claim);
432        assert!(score < 0.2, "low-ERL claim must be weighted down, got {score}");
433
434        // ECC DenyBarge → back to Speaking; no Mute/Cancel on the edge.
435        let (verdict, cmds) = s.resolve_overlap_from_erl(claim.erl);
436        assert_eq!(verdict, FloorVerdict::DenyBarge);
437        assert!(
438            !cmds.iter().any(|c| matches!(
439                c,
440                EdgeCommand::Mute | EdgeCommand::Cancel
441            )),
442            "self-cancel fix: answer not cancelled"
443        );
444        assert_eq!(s.duplex.state(), DuplexState::Speaking);
445        assert_eq!(s.edge.reflex(), EdgeReflexShadow::Streaming);
446    }
447
448    #[test]
449    fn genuine_barge_high_erl_mutes_edge() {
450        let mut s = LocalhostDuplexSession::open(PayloadKind::Audio);
451        let _ = s.daemon_observe(StreamObservation::VadOnset);
452        let _ = s.daemon_observe(StreamObservation::SmartTurnEou);
453        let _ = s.apply_verdict(FloorVerdict::GrantAgent);
454        assert_eq!(s.duplex.state(), DuplexState::Speaking);
455
456        s.edge.observe_erl(0.92);
457        s.edge_report_state();
458        let out = s.daemon_observe(StreamObservation::VadOnset);
459        assert_eq!(out.duplex_state, DuplexState::Overlap);
460        let claim = out.impulse.expect("TurnClaim");
461        assert!(erl_admits_barge(claim.erl.unwrap()));
462
463        let (verdict, cmds) = s.resolve_overlap_from_erl(claim.erl);
464        assert_eq!(verdict, FloorVerdict::GrantUser);
465        assert!(
466            cmds.iter().any(|c| matches!(c, EdgeCommand::Mute)),
467            "genuine barge must Mute the edge"
468        );
469        assert_eq!(s.duplex.state(), DuplexState::Listening);
470        assert_eq!(s.edge.reflex(), EdgeReflexShadow::Muted);
471    }
472
473    #[test]
474    fn media_up_only_while_transmitting() {
475        let mut s = LocalhostDuplexSession::open(PayloadKind::Audio);
476        s.edge_push_media(MediaFrame::Pcm(vec![1, 2, 3]));
477        assert_eq!(s.media_up.len(), 1);
478        s.edge.on_link_loss();
479        s.edge_push_media(MediaFrame::Pcm(vec![4, 5, 6]));
480        assert_eq!(s.media_up.len(), 1, "Degraded edge must not TX");
481    }
482
483    /// WEFT-615: duplex TurnClaim payload always carries erl_confidence so
484    /// TalkModeLoop's acoustic gate (not a binary flag) can fire.
485    #[test]
486    fn duplex_claim_payload_carries_erl() {
487        let with = DuplexImpulse {
488            kind: ImpulseType::TurnClaim,
489            erl: Some(0.88),
490        };
491        let p = LocalhostDuplexSession::duplex_turn_claim_payload(&with);
492        assert!((p["erl_confidence"].as_f64().unwrap() - 0.88).abs() < 1e-6);
493
494        let missing = DuplexImpulse {
495            kind: ImpulseType::TurnClaim,
496            erl: None,
497        };
498        let p0 = LocalhostDuplexSession::duplex_turn_claim_payload(&missing);
499        assert_eq!(p0["erl_confidence"].as_f64().unwrap(), 0.0);
500    }
501
502    /// WEFT-615: hysteresis gate holds deny/admit across the mid-band during
503    /// provisional Overlap (no Mute thrash on AEC residual flutter).
504    #[test]
505    fn overlap_gate_hysteresis_no_thrash() {
506        let mut s = LocalhostDuplexSession::open(PayloadKind::Audio);
507        let _ = s.daemon_observe(StreamObservation::VadOnset);
508        let _ = s.daemon_observe(StreamObservation::SmartTurnEou);
509        let _ = s.apply_verdict(FloorVerdict::GrantAgent);
510        // Enter Overlap.
511        let _ = s.daemon_observe(StreamObservation::VadOnset);
512        assert_eq!(s.duplex.state(), DuplexState::Overlap);
513
514        // Self-echo samples stay denied — no Mute.
515        let (v, cmds) = s.resolve_overlap_with_gate(Some(0.2));
516        assert_eq!(v, FloorVerdict::DenyBarge);
517        assert!(!cmds.iter().any(|c| matches!(c, EdgeCommand::Mute)));
518        assert_eq!(s.duplex.state(), DuplexState::Speaking);
519
520        // Re-enter Overlap for a genuine barge, then flutter in-band.
521        let _ = s.daemon_observe(StreamObservation::VadOnset);
522        let (v, _) = s.resolve_overlap_with_gate(Some(0.9));
523        assert_eq!(v, FloorVerdict::GrantUser);
524        assert_eq!(s.duplex.state(), DuplexState::Listening);
525
526        // After grant we're Listening; gate itself holds admit in-band.
527        assert!(s.erl_gate.observe(0.45));
528        assert!(!s.erl_gate.observe(0.2));
529    }
530
531    /// Custom floor knob: admit only above a stricter threshold.
532    #[test]
533    fn custom_erl_floor_knob() {
534        let mut s = LocalhostDuplexSession::open_with_erl(
535            PayloadKind::Audio,
536            ErlBargeConfig::new(0.8, 0.1),
537        );
538        let _ = s.daemon_observe(StreamObservation::VadOnset);
539        let _ = s.daemon_observe(StreamObservation::SmartTurnEou);
540        let _ = s.apply_verdict(FloorVerdict::GrantAgent);
541        let _ = s.daemon_observe(StreamObservation::VadOnset);
542        let (v, _) = s.resolve_overlap_with_config(Some(0.7), &ErlBargeConfig::new(0.8, 0.1));
543        assert_eq!(v, FloorVerdict::DenyBarge, "0.7 < custom floor 0.8");
544        // Re-overlap and clear the stricter floor.
545        let _ = s.daemon_observe(StreamObservation::VadOnset);
546        let (v, cmds) = s.resolve_overlap_with_config(Some(0.85), &ErlBargeConfig::new(0.8, 0.1));
547        assert_eq!(v, FloorVerdict::GrantUser);
548        assert!(cmds.iter().any(|c| matches!(c, EdgeCommand::Mute)));
549    }
550}
clawft_kernel/thin_edge.rs

clawft_kernel/
thin_edge.rs
