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

//! `wave51_streaming_trace_binding` — multi-window correlation across
//! the JouleClaw / Sandbox / Joule-Code triad.
//!
//! Composes waves 47 + 50 + 49 + sandbox into one binding:
//!
//! - **Wave-47** — N `stream-window/1` receipts, one per 200 ms
//!   window, each carrying the per-window energy reading.
//! - **Wave-50** — N matching `coding-window/1` receipts plus a
//!   materialized op_id when the session sealed accepted edits.
//! - **Sandbox** — one long-lived attestation wrapping the whole
//!   session at tier S5 with Cpu+Gpu grants.
//! - **Wave-51 (new)** — `StreamingTraceBinding` records all three
//!   under one trace_id; `StreamingInvariantVerifier` asserts:
//!     · sandbox.trace_id == binding.trace_id
//!     · every window's stream_id == binding.stream_id
//!     · window ids strictly ascending
//!     · Σ window.joules ≈ sandbox.joules (within tolerance)
//!     · declared_provenance == worst observed across windows + sandbox
//!     · session_op_id (if non-empty) accounted by a window's
//!       materialized_op_id
//!     · stewards trusted
//!
//! Run:
//!     cargo run -p jouleclaw-stack --example wave51_streaming_trace_binding

use jouleclaw_coding_stream::{
    CODING_STREAM_MIME, CodingAnchor, CodingEvent, CodingOpMaterializer, CodingWindow,
};
use jouleclaw_cross_binding::{
    EnergyProvenance, SandboxRef, StreamingInvariantVerifier, StreamingTraceBinding, TraceId,
    WindowEnergyRef,
};
use jouleclaw_jcr1::{CodingActivitySummary, CodingWindowReceipt};
use jouleclaw_stream::{
    ChannelDirection, ChannelSpec, StreamSpec, StreamWindow, Tick, TickKind, WindowId,
};
use std::fs;

const STEWARD: &str = "did:keri:EJ9testStewardAID000000000000000";
const SUPERVISOR: &str = "6e7a1cdd29b0b78fd13af4c5598feff4ef2a97166e3ca6f2e4fbfccd80505bf1";
const AUTHOR: &str = "d75a980182b10ab7d54bfed3c964073a0ee172f3daa62325af021a68f707511a";
const STREAM_ID: &str = "session-wave51";

fn coding_spec() -> StreamSpec {
    StreamSpec {
        tick_window_ms: 200,
        max_chunk_bytes: 64 * 1024,
        max_ticks_per_window: 64,
        channels: vec![ChannelSpec {
            id: "coding".into(),
            direction: ChannelDirection::Bidi,
            kind_tag: "text".into(),
            mime: CODING_STREAM_MIME.into(),
        }],
    }
}

fn tick(spec: &StreamSpec, w: u64, ev: CodingEvent, ts: u64) -> Tick {
    let t = Tick {
        window_id: WindowId(w),
        channel_id: "coding".into(),
        kind: TickKind::Text,
        bytes: ev.to_tick_bytes().unwrap(),
        ts_ns: ts,
    };
    spec.validate_tick(&t).unwrap();
    t
}

fn build_window(spec: &StreamSpec, w: u64, events: Vec<CodingEvent>) -> StreamWindow {
    let mut sw = StreamWindow::new(WindowId(w));
    for (i, ev) in events.into_iter().enumerate() {
        sw.push(spec, tick(spec, w, ev, (i as u64) * 1000)).unwrap();
    }
    sw
}

fn summary(cw: &CodingWindow) -> CodingActivitySummary {
    let mut delta = 0i32;
    for ev in &cw.events {
        if let CodingEvent::DiagnosticDelta { added, cleared, .. } = ev {
            delta += *added as i32 - *cleared as i32;
        }
    }
    CodingActivitySummary {
        typed_bytes: cw.typed_bytes() as u64,
        suggestions_appeared: cw.count_by_tag("suggestion_appeared") as u32,
        suggestions_accepted: cw.accepted_count() as u32,
        suggestions_rejected: cw.rejected_count() as u32,
        cursor_moves: cw.count_by_tag("cursor_moved") as u32,
        diagnostic_delta_net: delta,
    }
}

fn main() {
    println!("== wave-51: streaming-trace binding (multi-window, three standards) ==\n");

    let trace_id = TraceId::new("b3:wave51-trace-0001-wave51-trace-0001-wave51000000");
    let spec = coding_spec();

    // Scripted streaming coding session: 4 windows.
    let session: Vec<(u64, u64, Vec<CodingEvent>)> = vec![
        (
            0,
            350_000, // µJ for this window
            vec![CodingEvent::TypedChar {
                path: "src/lib.rs".into(),
                bytes: "fn add".into(),
            }],
        ),
        (
            1,
            420_000,
            vec![
                CodingEvent::TypedChar {
                    path: "src/lib.rs".into(),
                    bytes: "(a,b)".into(),
                },
                CodingEvent::SuggestionAppeared {
                    suggestion_id: "b3:sugg-0001-fill-out-signature-and-body".into(),
                    path: "src/lib.rs".into(),
                    anchor: CodingAnchor::ByteRange {
                        start: 0,
                        end: 12,
                    },
                    replacement: "pub fn add(a: i64, b: i64) -> i64 { a + b }".into(),
                },
            ],
        ),
        (
            2,
            330_000,
            vec![
                CodingEvent::SuggestionAccepted {
                    suggestion_id: "b3:sugg-0001-fill-out-signature-and-body".into(),
                },
                CodingEvent::CursorMoved {
                    path: "src/lib.rs".into(),
                    offset: 43,
                },
            ],
        ),
        (
            3,
            280_000,
            vec![CodingEvent::DiagnosticDelta {
                path: "src/lib.rs".into(),
                added: 0,
                cleared: 2,
            }],
        ),
    ];

    let mut materializer = CodingOpMaterializer::new(
        "b3:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
        AUTHOR,
        "L3",
    );

    let mut window_refs: Vec<WindowEnergyRef> = Vec::new();
    let mut total_joules: u64 = 0;
    let mut session_op_id = String::new();

    println!("== per-window summary ==");
    for (wid, joules, events) in &session {
        let sw = build_window(&spec, *wid, events.clone());
        let cw = CodingWindow::from_window(&sw);
        materializer.observe_window(&cw);

        // Materialize at the LAST window with accepts.
        let materialized_op_id = if *wid == session.last().unwrap().0 {
            match materializer.materialize("add(a,b)->i64") {
                Ok(op) => {
                    session_op_id = op.op_id.clone();
                    op.op_id
                }
                Err(_) => String::new(),
            }
        } else {
            String::new()
        };

        let activity = summary(&cw);
        let coding_receipt = CodingWindowReceipt::build(
            format!("{STREAM_ID}:cw:{:04}", wid),
            format!("2026-06-05T00:00:{:02}Z", wid),
            STEWARD,
            STREAM_ID,
            *wid,
            activity.clone(),
            materialized_op_id.clone(),
            if materialized_op_id.is_empty() {
                ""
            } else {
                AUTHOR
            },
        );
        let coding_hash = coding_receipt.receipt_hash().unwrap();
        let stream_receipt_id = format!("{STREAM_ID}:sw:{:04}", wid);

        window_refs.push(WindowEnergyRef {
            stream_id: STREAM_ID.into(),
            window_id: *wid,
            stream_receipt_id: stream_receipt_id.clone(),
            coding_receipt_id: coding_receipt.id.clone(),
            joules_uj: *joules,
            energy_provenance: EnergyProvenance::Estimator,
            materialized_op_id: materialized_op_id.clone(),
        });
        total_joules += joules;

        let tag = if !materialized_op_id.is_empty() {
            "SEAL"
        } else if activity.suggestions_accepted > 0 {
            "ACPT"
        } else if activity.suggestions_appeared > 0 {
            "APPR"
        } else if activity.typed_bytes > 0 {
            "TYPE"
        } else {
            "IDLE"
        };
        println!(
            "  w{:02}  {} typed={:>3}B  appeared={} accepted={}  J={:>5}µJ  coding-receipt={}",
            wid,
            tag,
            activity.typed_bytes,
            activity.suggestions_appeared,
            activity.suggestions_accepted,
            joules,
            &coding_hash[..16]
        );
        if !materialized_op_id.is_empty() {
            println!("         ↳ materialized op_id={}…", &materialized_op_id[..20]);
        }
    }

    println!();
    println!("== sandbox attestation wrapping the session ==");
    let sandbox = SandboxRef {
        receipt_id: "b3:sandbox-wave51-receipt-0001-0001-0001-rcp00000".into(),
        content_id: "b3:sandbox-wave51-content-id-0001-0001-cid000000000".into(),
        tier: "S5".into(),
        granted_device_classes: vec!["Cpu".into(), "Gpu".into()],
        joules_uj: total_joules, // sandbox accounts the same total
        energy_provenance: EnergyProvenance::Estimator,
        supervisor_pubkey_hex: SUPERVISOR.into(),
        trace_id: trace_id.clone(),
    };
    println!(
        "  tier={}  grants={:?}  total={}µJ  supervisor={}…",
        sandbox.tier,
        sandbox.granted_device_classes,
        sandbox.joules_uj,
        &sandbox.supervisor_pubkey_hex[..16]
    );
    println!();

    // Build + verify the streaming-trace binding.
    let binding = StreamingTraceBinding::build(
        "stb-wave51-0001",
        "2026-06-05T00:00:04Z",
        STEWARD,
        trace_id.clone(),
        STREAM_ID,
        window_refs,
        sandbox,
        session_op_id.clone(),
        EnergyProvenance::Estimator,
        0,
    );
    let verifier = StreamingInvariantVerifier::new(vec![
        STEWARD.into(),
        SUPERVISOR.into(),
    ]);
    println!("== invariant verification ==");
    match verifier.verify(&binding) {
        Ok(()) => println!("  ✓ all streaming invariants hold"),
        Err(e) => panic!("invariant verification failed: {e}"),
    }
    let binding_hash = binding.content_hash().expect("hash");
    let binding_canonical = binding.canonical_bytes().expect("canonical");
    println!("  binding hash:    {binding_hash}");
    println!("  canonical:       {} bytes", binding_canonical.len());
    println!("  windows bound:   {}", binding.windows.len());
    println!("  Σ joules:        {}µJ", binding.total_joules_uj());
    println!("  worst observed:  {:?}", binding.observed_worst_provenance());
    println!(
        "  session op_id:   {}",
        if session_op_id.is_empty() {
            "(none)"
        } else {
            &session_op_id
        }
    );
    println!();

    // Emit the binding doc for downstream auditors.
    let doc = serde_json::from_slice::<serde_json::Value>(&binding_canonical).unwrap();
    fs::write(
        "/tmp/wave51-streaming-trace-binding.json",
        serde_json::to_string_pretty(&doc).unwrap(),
    )
    .expect("write");
    println!("  ✓ wrote /tmp/wave51-streaming-trace-binding.json");
    println!();
    println!(
        "✓ Multi-window streaming session, three open standards, one signed binding —"
    );
    println!("    StreamSpec + CodingEvents (wave-47 × wave-50)");
    println!("    → N stream-window/1 receipts + N coding-window/1 receipts");
    println!("    + 1 sandbox attestation + 0..1 materialized joule-code op");
    println!("    → StreamingTraceBinding (wave-51) ties them under one trace_id");
    println!("      with verifiable invariants no per-standard verifier could enforce alone.");
}