open-standards/jouleclaw/jouleclaw-rs/crates/jouleclaw-stack/examples/wave47_streaming_pipeline.rs

//! `wave47_streaming_pipeline` — streaming-cascade substrate demo.
//!
//! Simulates a duplex-voice stream of audio + marker ticks over
//! 13 windows (~2.6 s at 200 ms/window). For each window the
//! pipeline runs:
//!
//! 1. **Stream Oracle** (CompositeStreamOracle = always-pass +
//!    barge-in detector + per-tick adapter wrapping the no-`unsafe`
//!    fuzz property on the text channel) — emits a [`WindowVerdict`].
//!    Dominance order: Fail > Yield > NeedMore > Pass.
//! 2. **Rolling conservation ledger** (4-window buffer, 256 B
//!    tolerance) — emits a [`WindowConservationFinding`]. The
//!    barge-in deadline is 1 window; if the agent keeps emitting
//!    outbound past it, the ledger flags `BargeInIgnored`.
//! 3. **JCR-1 `stream-window/1` receipt** — minted per window with
//!    verdict + conservation + joules. Canonical JSON, BLAKE3-
//!    addressed, ready for COSE_Sign1.
//!
//! Run:
//!     cargo run -p jouleclaw-stack --example wave47_streaming_pipeline

use jouleclaw_jcr1::{
    EnergyProvenanceWire, StreamWindowReceipt, WindowConservationWire, WindowVerdictWire,
};
use jouleclaw_oracle::AlwaysPassOracle;
use jouleclaw_stream::{
    BARGE_IN_DEADLINE_MS, ChannelDirection, ChannelSpec, DEFAULT_TICK_WINDOW_MS,
    RollingConservationLedger, StreamSpec, StreamWindow, Tick, TickMarker, WindowConservationFinding,
    WindowId,
};
use jouleclaw_stream_oracle::{
    AlwaysPassStream, BargeInDetector, CompositeStreamOracle, PerTickAdapter, StreamOracle,
    WindowVerdict,
};

const STEWARD: &str = "did:keri:EJ9testStewardAID000000000000000";
const STREAM_ID: &str = "demo-duplex";
const CLOSED_AT_BASE: u64 = 1_780_000_000;

fn duplex_spec_with_text() -> StreamSpec {
    let mut base = StreamSpec::duplex_voice_default();
    // Add an outbound text channel so we can demo the per-tick
    // adapter wrapping the no-`unsafe` property.
    base.channels.push(ChannelSpec {
        id: "text.out".into(),
        direction: ChannelDirection::Outbound,
        kind_tag: "text".into(),
        mime: "text/plain;charset=utf-8".into(),
    });
    base
}

/// One scripted window of the simulated stream.
struct ScriptedWindow {
    id: u64,
    inbound_b: usize,
    outbound_b: usize,
    text_out: Option<&'static str>,
    marker: Option<TickMarker>,
    joules_uj: u64,
}

fn build_window(spec: &StreamSpec, w: &ScriptedWindow) -> StreamWindow {
    let id = WindowId(w.id);
    let mut window = StreamWindow::new(id);
    if w.inbound_b > 0 {
        window
            .push(spec, Tick::audio(id, "audio.in", vec![0u8; w.inbound_b], 0))
            .expect("inbound");
    }
    if w.outbound_b > 0 {
        window
            .push(
                spec,
                Tick::audio(id, "audio.out", vec![0u8; w.outbound_b], 100),
            )
            .expect("outbound");
    }
    if let Some(t) = w.text_out {
        window
            .push(spec, Tick::text(id, "text.out", t, 150))
            .expect("text");
    }
    if let Some(m) = &w.marker {
        window
            .push(spec, Tick::marker(id, "markers", m.clone(), 200))
            .expect("marker");
    }
    window
}

fn verdict_to_wire(v: &WindowVerdict) -> WindowVerdictWire {
    match v {
        WindowVerdict::Pass { chain } => WindowVerdictWire::Pass {
            chain: chain.clone(),
        },
        WindowVerdict::Fail { reasons } => WindowVerdictWire::Fail {
            reasons: reasons
                .iter()
                .map(|r| serde_json::to_string(r).unwrap_or_else(|_| "fail".into()))
                .collect(),
        },
        WindowVerdict::Yield {
            reason: jouleclaw_stream_oracle::WindowFailReason::BargeIn { seen_at },
        } => WindowVerdictWire::Yield {
            seen_at: seen_at.0,
        },
        WindowVerdict::Yield { reason } => WindowVerdictWire::Yield {
            seen_at: match reason {
                jouleclaw_stream_oracle::WindowFailReason::BargeIn { seen_at } => seen_at.0,
                _ => 0,
            },
        },
        WindowVerdict::NeedMore { detail } => WindowVerdictWire::NeedMore {
            detail: detail.clone(),
        },
    }
}

fn conservation_to_wire(f: &WindowConservationFinding) -> WindowConservationWire {
    match f {
        WindowConservationFinding::Conforming => WindowConservationWire::Conforming,
        WindowConservationFinding::Hallucination { unaccounted } => {
            WindowConservationWire::Hallucination {
                unaccounted: *unaccounted as u64,
            }
        }
        WindowConservationFinding::BargeInIgnored { fired_at } => {
            WindowConservationWire::BargeInIgnored {
                fired_at: fired_at.0,
            }
        }
    }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    println!(
        "== wave-47 streaming pipeline demo  (tick={} ms, barge-in deadline={} ms) ==\n",
        DEFAULT_TICK_WINDOW_MS, BARGE_IN_DEADLINE_MS
    );

    let spec = duplex_spec_with_text();

    // Stream Oracle: always-pass + barge-in detector + per-tick
    // adapter wrapping a "no-`unsafe`" check on the text channel.
    let oracle = CompositeStreamOracle::new("composite-stream")
        .add(Box::new(AlwaysPassStream))
        .add(Box::new(BargeInDetector))
        .add(Box::new(PerTickAdapter::new(
            "no-unsafe-on-text",
            AlwaysPassOracle, // stub — a real Oracle would scan tick bytes
            "text".to_string(),
            "text.out".to_string(),
        )));

    let mut ledger = RollingConservationLedger::new(4, 256)
        .with_barge_in_deadline_windows(1);

    let script = vec![
        // 0..5: clean conversation, agent emits short replies.
        ScriptedWindow {
            id: 0,
            inbound_b: 6400,
            outbound_b: 0,
            text_out: None,
            marker: None,
            joules_uj: 80,
        },
        ScriptedWindow {
            id: 1,
            inbound_b: 6400,
            outbound_b: 0,
            text_out: None,
            marker: None,
            joules_uj: 90,
        },
        ScriptedWindow {
            id: 2,
            inbound_b: 6400,
            outbound_b: 0,
            text_out: None,
            marker: Some(TickMarker::EndOfTurn),
            joules_uj: 95,
        },
        ScriptedWindow {
            id: 3,
            inbound_b: 0,
            outbound_b: 4800,
            text_out: Some("hello, what can I help with?"),
            marker: None,
            joules_uj: 340,
        },
        ScriptedWindow {
            id: 4,
            inbound_b: 0,
            outbound_b: 4800,
            text_out: None,
            marker: None,
            joules_uj: 320,
        },
        ScriptedWindow {
            id: 5,
            inbound_b: 0,
            outbound_b: 4800,
            text_out: None,
            marker: None,
            joules_uj: 320,
        },
        // 6: barge-in fires.
        ScriptedWindow {
            id: 6,
            inbound_b: 1600,
            outbound_b: 4800,
            text_out: None,
            marker: Some(TickMarker::BargeIn),
            joules_uj: 280,
        },
        // 7: well-behaved cascade yields — outbound drops to zero.
        ScriptedWindow {
            id: 7,
            inbound_b: 6400,
            outbound_b: 0,
            text_out: None,
            marker: None,
            joules_uj: 60,
        },
        // 8..9: user continues, agent listens.
        ScriptedWindow {
            id: 8,
            inbound_b: 6400,
            outbound_b: 0,
            text_out: None,
            marker: None,
            joules_uj: 70,
        },
        ScriptedWindow {
            id: 9,
            inbound_b: 6400,
            outbound_b: 0,
            text_out: None,
            marker: Some(TickMarker::EndOfTurn),
            joules_uj: 75,
        },
        // 10: agent reply.
        ScriptedWindow {
            id: 10,
            inbound_b: 0,
            outbound_b: 4800,
            text_out: Some("got it, let me think."),
            marker: None,
            joules_uj: 340,
        },
        // 11: HALLUCINATION — outbound 32 KiB with no inbound, no
        // substrate. The rolling ledger flags it.
        ScriptedWindow {
            id: 11,
            inbound_b: 0,
            outbound_b: 32 * 1024,
            text_out: None,
            marker: None,
            joules_uj: 1200,
        },
        ScriptedWindow {
            id: 12,
            inbound_b: 0,
            outbound_b: 0,
            text_out: None,
            marker: Some(TickMarker::End),
            joules_uj: 20,
        },
    ];

    let mut pass = 0usize;
    let mut yield_ = 0usize;
    let mut fail = 0usize;
    let mut hallucination = 0usize;
    let mut total_uj: u64 = 0;

    for sw in &script {
        let window = build_window(&spec, sw);
        let verdict = oracle.verify_window(&window).await.expect("verify");
        let finding = ledger.tick(&window, &spec, /* substrate_bytes */ 0);
        total_uj += sw.joules_uj;
        match &verdict {
            WindowVerdict::Pass { .. } => pass += 1,
            WindowVerdict::Yield { .. } => yield_ += 1,
            WindowVerdict::Fail { .. } => fail += 1,
            WindowVerdict::NeedMore { .. } => {}
        }
        if matches!(finding, WindowConservationFinding::Hallucination { .. }) {
            hallucination += 1;
        }

        // Mint the receipt.
        let receipt = StreamWindowReceipt::build(
            format!("{}:{:04}", STREAM_ID, sw.id),
            format!(
                "2026-06-05T00:00:{:02}Z",
                CLOSED_AT_BASE.wrapping_add(sw.id) % 60
            ),
            STEWARD,
            STREAM_ID,
            sw.id,
            window.content_hash(),
            window.ticks.len() as u64,
            window.inbound_bytes(&spec) as u64,
            window.outbound_bytes(&spec) as u64,
            sw.joules_uj,
            EnergyProvenanceWire::Estimator,
            verdict_to_wire(&verdict),
            conservation_to_wire(&finding),
        );

        // Sanity-check canonical encoding + content hash.
        let canonical = receipt.canonical_bytes().expect("canonical");
        let receipt_hash = receipt.content_hash().expect("hash");

        let tag = match (&verdict, &finding) {
            (WindowVerdict::Yield { .. }, _) => "YIELD",
            (WindowVerdict::Fail { .. }, _) => "FAIL ",
            (_, WindowConservationFinding::Hallucination { .. }) => "HALLU",
            (_, WindowConservationFinding::BargeInIgnored { .. }) => "IGNRD",
            _ => "PASS ",
        };
        println!(
            "  w{:02}  {} ticks={} in={:>5}B out={:>6}B  J={:>5}µJ  [{}]  hash={}",
            sw.id,
            tag,
            window.ticks.len(),
            window.inbound_bytes(&spec),
            window.outbound_bytes(&spec),
            sw.joules_uj,
            short(&receipt_hash),
            short(&hex_short(&canonical)),
        );
    }

    println!();
    println!("== summary ==");
    println!(
        "  windows: {}  |  pass={}  yield={}  fail={}  hallucination={}",
        script.len(),
        pass,
        yield_,
        fail,
        hallucination
    );
    println!("  total joules: {} µJ", total_uj);
    println!();
    println!("✓ Streaming cascade substrate works end-to-end:");
    println!("    StreamSpec → StreamWindow → CompositeStreamOracle (Pass | Yield | Fail | NeedMore)");
    println!("    → RollingConservationLedger (Conforming | Hallucination | BargeInIgnored)");
    println!("    → JCR-1 stream-window/1 receipt per window");
}

fn short(s: &str) -> String {
    if s.len() <= 12 {
        s.to_string()
    } else {
        format!("{}…", &s[..12])
    }
}

fn hex_short(bytes: &[u8]) -> String {
    let mut s = String::new();
    for b in bytes.iter().take(6) {
        use std::fmt::Write;
        let _ = write!(s, "{b:02x}");
    }
    s
}