diff --git a/contracts/spec/gen-rust.ts b/contracts/spec/gen-rust.ts index 78e23564..0e3bd3cd 100644 --- a/contracts/spec/gen-rust.ts +++ b/contracts/spec/gen-rust.ts @@ -72,6 +72,19 @@ import { STREAM_VRING_OFF, SVC_IMG_MAX_BYTES, SVC_POLL_BUF, + WIRE_BEACON_MAGIC, + WIRE_BEACON_PORT, + WIRE_CHUNK_HEADER_SIZE, + WIRE_HEADER_SIZE, + WIRE_MAGIC, + WIRE_MARK_FLAG_ENDED, + WIRE_MARK_SIZE, + WIRE_MAX_PAYLOAD, + WIRE_MSG, + WIRE_PORT, + WIRE_SLOT_FLAG_RLE, + WIRE_SLOT_HEADER_SIZE, + WIRE_VERSION, TEX_MAX_DIM, TEX_SLOT_BITS, TEX_SLOT_MASK, @@ -318,6 +331,26 @@ export function generateRust(): string { put(` pub const FLAG_ENDED: u16 = ${STREAM_FLAG_ENDED};`); put("}"); put(""); + put("/// SVC WIRE protocol (PKNT) — the svc mailbox over a socket."); + put("/// Full byte layout in spec.ts; parsed by engine/core/src/wire.rs."); + put("pub mod wire {"); + put(` pub const MAGIC: u32 = ${hex(WIRE_MAGIC)}; // 'PKNT' LE`); + put(` pub const BEACON_MAGIC: u32 = ${hex(WIRE_BEACON_MAGIC)}; // 'PKDB' LE`); + put(` pub const VERSION: u8 = ${WIRE_VERSION};`); + put(` pub const HEADER_SIZE: usize = ${WIRE_HEADER_SIZE};`); + put(` pub const MAX_PAYLOAD: usize = ${WIRE_MAX_PAYLOAD};`); + put(` pub const BEACON_PORT: u16 = ${WIRE_BEACON_PORT};`); + put(` pub const PORT: u16 = ${WIRE_PORT};`); + for (const [name, v] of Object.entries(WIRE_MSG)) { + put(` pub const MSG_${screaming(name)}: u8 = ${hex(v, 2)};`); + } + put(` pub const SLOT_HEADER_SIZE: usize = ${WIRE_SLOT_HEADER_SIZE};`); + put(` pub const CHUNK_HEADER_SIZE: usize = ${WIRE_CHUNK_HEADER_SIZE};`); + put(` pub const MARK_SIZE: usize = ${WIRE_MARK_SIZE};`); + put(` pub const SLOT_FLAG_RLE: u16 = ${WIRE_SLOT_FLAG_RLE};`); + put(` pub const MARK_FLAG_ENDED: u16 = ${WIRE_MARK_FLAG_ENDED};`); + put("}"); + put(""); // --- style table -------------------------------------------------------------- put("/// STYLE TABLE (styles.bin) format constants — full layout in spec.ts."); diff --git a/contracts/spec/spec.ts b/contracts/spec/spec.ts index 11f32bb6..245f958d 100644 --- a/contracts/spec/spec.ts +++ b/contracts/spec/spec.ts @@ -737,6 +737,95 @@ export const STREAM_SLOT_HEADER_SIZE = 32; export const STREAM_CHUNK_HEADER_SIZE = 16; export const STREAM_FLAG_ENDED = 1 << 0; +// --------------------------------------------------------------------------- +// SVC WIRE protocol (PKNT) — the svc mailbox over a socket +// --------------------------------------------------------------------------- +// The PSP reaches its companion host through PSPLINK's usbhostfs file share +// (SVC above); hosts without a shared filesystem (the Vita, over WiFi) speak +// the SAME mailbox + side-file + .pkst semantics over one TCP connection. +// One connection on purpose: TCP ordering is load-bearing — STREAM_OPEN +// precedes the JSON line announcing it, FILE pushes precede the results line +// that references them — and a second channel would reintroduce exactly the +// cross-channel races the file transport never had. +// +// Discovery: the host broadcasts a UDP beacon once a second on +// WIRE_BEACON_PORT; the datagram's SOURCE ADDRESS is the connect target. +// +// Beacon datagram: +// off 0 u32 magic = 0x42444b50 bytes 'P','K','D','B' +// off 4 u8 version = WIRE_VERSION +// off 5 u8 reserved (0) +// off 6 u16 tcpPort the WIRE listener's port +// off 8 u8 appLen, then app id bytes (the pocket-svc app segment) +// ... u8 nameLen, then a display name (<= 32 bytes, for pickers) +// +// Handshake (before any frame): +// device -> host: u32 magic 'PKNT' · u8 version · u8 reserved · +// u8 appLen · app id bytes +// host -> device: u32 magic 'PKNT' · u8 acceptedVersion · u8 flags (0) · +// u16 reserved +// A magic/version/app mismatch closes the socket. +// +// Every subsequent message, both directions, is one frame: +// +// Frame header (WIRE_HEADER_SIZE = 8 bytes): +// off 0 u8 type (WIRE_MSG) +// off 1 u8 flags type-specific (see below) +// off 2 u16 reserved (0) +// off 4 u32 payloadLen (<= WIRE_MAX_PAYLOAD; oversize closes the socket) +// +// Types (unknown types are skipped by length — forward compatible): +// ping (h->d) u32 token, every ~2 s; either side drops the +// pong (d->h) connection after ~10 s of silence. Pong echoes. +// ctrl (both) one UTF-8 JSON line, no trailing \n — the in.jsonl / +// out.jsonl mailbox semantics (<= SVC_POLL_BUF bytes). +// file (h->d) u16 pathLen · path (svc-relative, side_path rules) · +// whole IMG-entry bytes (<= SVC_IMG_MAX_BYTES). Pushed +// PROACTIVELY before the ctrl line that references it, +// so a synchronous loadImgFile hits a warm device cache. +// streamOpen (h->d) u16 pathLen · path (what the announce line will name) +// · the verbatim 96-byte .pkst header block. The device +// allocates a RAM ring image (stream_rx.rs) — every +// streamOpen is a full ring reset. +// streamClose (h->d) empty payload. +// videoSlot (h->d) u32 seq · u32 frameIndex · u16 w · u16 h · +// u16 flags (bit 0 = indices PackBits-RLE) · u16 rsv · +// u8[1024] palette · indices (raw w*h, or RLE decoding +// to exactly w*h). Applied payload-first, seq-published- +// after into the RAM ring, preserving the .pkst torn- +// frame contract verbatim. +// audioChunk (h->d) u32 seq · u32 startFrame · s16 PCM, exactly +// chunkFrames*channels samples. +// streamMark (h->d) u32 epoch · u16 flags (bit 0 = ended) · u16 reserved — +// carries bumpEpoch()/markEnded() into the ring header. + +export const WIRE_MAGIC = 0x544e4b50; // 'PKNT' LE +export const WIRE_BEACON_MAGIC = 0x42444b50; // 'PKDB' LE +export const WIRE_VERSION = 1; +export const WIRE_HEADER_SIZE = 8; +export const WIRE_MAX_PAYLOAD = 256 * 1024; +/** UDP discovery beacon (host broadcasts; device listens). */ +export const WIRE_BEACON_PORT = 8621; +/** The host's TCP listener for the framed protocol. */ +export const WIRE_PORT = 8622; +export const WIRE_MSG = { + ping: 0x01, + pong: 0x02, + ctrl: 0x10, + file: 0x20, + streamOpen: 0x30, + streamClose: 0x31, + videoSlot: 0x32, + audioChunk: 0x33, + streamMark: 0x34, +} as const; +/** videoSlot leading fields (seq..reserved) — the palette follows. */ +export const WIRE_SLOT_HEADER_SIZE = 16; +export const WIRE_CHUNK_HEADER_SIZE = 8; +export const WIRE_MARK_SIZE = 8; +export const WIRE_SLOT_FLAG_RLE = 1 << 0; +export const WIRE_MARK_FLAG_ENDED = 1 << 0; + // --------------------------------------------------------------------------- // Font slots // --------------------------------------------------------------------------- diff --git a/engine/core/src/lib.rs b/engine/core/src/lib.rs index a7514252..8b5be79a 100644 --- a/engine/core/src/lib.rs +++ b/engine/core/src/lib.rs @@ -38,9 +38,11 @@ pub mod package; pub mod raster; pub mod spec; pub mod stream; +pub mod stream_rx; pub mod style; pub mod text; pub mod tree; +pub mod wire; pub use draw::DrawList; diff --git a/engine/core/src/spec.rs b/engine/core/src/spec.rs index 6d19cc6c..1caa44a7 100644 --- a/engine/core/src/spec.rs +++ b/engine/core/src/spec.rs @@ -361,6 +361,32 @@ pub mod stream { pub const FLAG_ENDED: u16 = 1; } +/// SVC WIRE protocol (PKNT) — the svc mailbox over a socket. +/// Full byte layout in spec.ts; parsed by engine/core/src/wire.rs. +pub mod wire { + pub const MAGIC: u32 = 0x544e4b50; // 'PKNT' LE + pub const BEACON_MAGIC: u32 = 0x42444b50; // 'PKDB' LE + pub const VERSION: u8 = 1; + pub const HEADER_SIZE: usize = 8; + pub const MAX_PAYLOAD: usize = 262144; + pub const BEACON_PORT: u16 = 8621; + pub const PORT: u16 = 8622; + pub const MSG_PING: u8 = 0x01; + pub const MSG_PONG: u8 = 0x02; + pub const MSG_CTRL: u8 = 0x10; + pub const MSG_FILE: u8 = 0x20; + pub const MSG_STREAM_OPEN: u8 = 0x30; + pub const MSG_STREAM_CLOSE: u8 = 0x31; + pub const MSG_VIDEO_SLOT: u8 = 0x32; + pub const MSG_AUDIO_CHUNK: u8 = 0x33; + pub const MSG_STREAM_MARK: u8 = 0x34; + pub const SLOT_HEADER_SIZE: usize = 16; + pub const CHUNK_HEADER_SIZE: usize = 8; + pub const MARK_SIZE: usize = 8; + pub const SLOT_FLAG_RLE: u16 = 1; + pub const MARK_FLAG_ENDED: u16 = 1; +} + /// STYLE TABLE (styles.bin) format constants — full layout in spec.ts. pub mod style_table { pub const MAGIC: u32 = 0x54534344; // 'DCST' LE diff --git a/engine/core/src/stream_rx.rs b/engine/core/src/stream_rx.rs new file mode 100644 index 00000000..62854406 --- /dev/null +++ b/engine/core/src/stream_rx.rs @@ -0,0 +1,165 @@ +//! RAM `.pkst` assembler — the receive side of the WIRE stream messages. +//! +//! `RamStream` maintains a byte-exact `.pkst` FILE IMAGE in memory: the same +//! 96-byte header block, the same video/audio rings at the same offsets. A +//! host that receives WIRE videoSlot/audioChunk/streamMark messages applies +//! them here, then points the exact same reader logic (stream.rs +//! parse_header_block / slot_offset / parse_slot_header / chunk_offset) at +//! `buf()` — the file-backed and socket-backed transports share one reader, +//! one torn-frame contract, one golden format. +//! +//! The publish order is preserved verbatim: payload bytes land first, the +//! ring's latestSeq is written after. Over TCP nothing can tear mid-slot, +//! but the reader's seq re-check stays meaningful across epoch resets and +//! writer laps, and keeping the invariant means the two transports are +//! byte-equivalent, not just morally similar. + +use alloc::vec; +use alloc::vec::Vec; + +use crate::codec; +use crate::spec::stream as st; +use crate::spec::wire; +use crate::stream::{self, StreamHeaders}; + +pub struct RamStream { + buf: Vec, + /// Ring GEOMETRY parsed at open (fixed for the stream's lifetime). Live + /// cursors (latestSeq, epoch, ended) are read from `buf` like any reader. + geo: StreamHeaders, +} + +impl RamStream { + /// Allocate the ring image from a verbatim 96-byte header block (a WIRE + /// streamOpen payload). The block is validated by the same + /// parse_header_block every reader runs; a malformed one is refused here, + /// before any allocation is trusted. + pub fn open(header_block: &[u8]) -> Option { + let geo = stream::parse_header_block(header_block)?; + let chunk_size = stream::chunk_size(geo.audio.chunk_frames, geo.audio.channels)?; + let video_end = geo.video_off.checked_add(geo.video.slot_count.checked_mul(geo.video.slot_size)?)?; + let audio_end = geo.audio_off.checked_add(geo.audio.chunk_count.checked_mul(chunk_size)?)?; + let total = video_end.max(audio_end).max(st::HEADER_BLOCK_SIZE as u32) as usize; + let mut buf = vec![0u8; total]; + buf[..st::HEADER_BLOCK_SIZE].copy_from_slice(&header_block[..st::HEADER_BLOCK_SIZE]); + Some(Self { buf, geo }) + } + + /// The `.pkst` file image — hand it to the stream.rs readers. + pub fn buf(&self) -> &[u8] { + &self.buf + } + + pub fn geometry(&self) -> &StreamHeaders { + &self.geo + } + + fn wr_u32(&mut self, off: usize, v: u32) { + self.buf[off..off + 4].copy_from_slice(&v.to_le_bytes()); + } + + fn wr_u16(&mut self, off: usize, v: u16) { + self.buf[off..off + 2].copy_from_slice(&v.to_le_bytes()); + } + + /// Apply a WIRE videoSlot: write the slot payload, THEN publish the + /// ring's latestSeq. Rejects geometry that disagrees with the open + /// header, palettes that are not 1024 bytes, and indices that are not + /// (or do not RLE-decode to) exactly w*h. + pub fn apply_slot(&mut self, msg: &crate::wire::VideoSlotMsg) -> bool { + let v = self.geo.video; + if msg.w != v.w || msg.h != v.h || msg.palette.len() != 1024 { + return false; + } + let pixels = (v.w * v.h) as usize; + let Some(off) = stream::slot_offset(&self.geo, msg.seq) else { + return false; + }; + let off = off as usize; + let indices_at = off + st::SLOT_HEADER_SIZE + 1024; + if indices_at + pixels > self.buf.len() { + return false; + } + if msg.rle { + if !codec::packbits_decode(msg.indices, &mut self.buf[indices_at..indices_at + pixels]) { + return false; + } + } else { + if msg.indices.len() != pixels { + return false; + } + self.buf[indices_at..indices_at + pixels].copy_from_slice(msg.indices); + } + self.buf[off + st::SLOT_HEADER_SIZE..indices_at].copy_from_slice(msg.palette); + // Slot header: seq · frameIndex · w · h · reserved(20B zeros). + self.wr_u32(off, msg.seq); + self.wr_u32(off + 4, msg.frame_index); + self.wr_u16(off + 8, msg.w as u16); + self.wr_u16(off + 10, msg.h as u16); + for b in &mut self.buf[off + 12..off + st::SLOT_HEADER_SIZE] { + *b = 0; + } + // Payload is in place — NOW publish the cursor. + self.wr_u32(st::VRING_OFF + 20, msg.seq.max(self.latest_video_seq())); + true + } + + /// Apply a WIRE audioChunk: payload first, latestSeq after. + pub fn apply_chunk(&mut self, msg: &crate::wire::AudioChunkMsg) -> bool { + let a = self.geo.audio; + let bytes = (a.chunk_frames * a.channels * 2) as usize; + if msg.pcm.len() != bytes { + return false; + } + let Some(off) = stream::chunk_offset(&self.geo, msg.seq) else { + return false; + }; + let off = off as usize; + let pcm_at = off + st::CHUNK_HEADER_SIZE; + if pcm_at + bytes > self.buf.len() { + return false; + } + self.buf[pcm_at..pcm_at + bytes].copy_from_slice(msg.pcm); + self.wr_u32(off, msg.seq); + self.wr_u32(off + 4, msg.start_frame); + for b in &mut self.buf[off + 8..off + st::CHUNK_HEADER_SIZE] { + *b = 0; + } + self.wr_u32(st::ARING_OFF + 20, msg.seq.max(self.latest_audio_seq())); + true + } + + /// Apply a WIRE streamMark: epoch bump and/or the ended flag — the same + /// header fields the file writer's bumpEpoch()/markEnded() mutate. + pub fn apply_mark(&mut self, msg: &crate::wire::StreamMarkMsg) { + self.wr_u32(8, msg.epoch); + let mut flags = u16::from_le_bytes([self.buf[6], self.buf[7]]); + if msg.ended { + flags |= st::FLAG_ENDED; + } + self.wr_u16(6, flags); + } + + fn latest_video_seq(&self) -> u32 { + u32::from_le_bytes([ + self.buf[st::VRING_OFF + 20], + self.buf[st::VRING_OFF + 21], + self.buf[st::VRING_OFF + 22], + self.buf[st::VRING_OFF + 23], + ]) + } + + fn latest_audio_seq(&self) -> u32 { + u32::from_le_bytes([ + self.buf[st::ARING_OFF + 20], + self.buf[st::ARING_OFF + 21], + self.buf[st::ARING_OFF + 22], + self.buf[st::ARING_OFF + 23], + ]) + } + + const _ASSERT_MSG_SIZES: () = { + assert!(wire::SLOT_HEADER_SIZE == 16); + assert!(wire::CHUNK_HEADER_SIZE == 8); + }; +} diff --git a/engine/core/src/tests.rs b/engine/core/src/tests.rs index 39151ce2..d57bc727 100644 --- a/engine/core/src/tests.rs +++ b/engine/core/src/tests.rs @@ -2802,3 +2802,342 @@ fn stream_golden_fixture_parses() { let plane = ui.upload_texture(&init, 16, 16, spec::psm::PSM_T8); assert!(ui.update_texture_t8(plane, pal, px)); } + +// ---- SVC WIRE (PKNT) — wire.rs codec + stream_rx.rs RAM ring --------------- + +#[test] +fn wire_frame_header_round_trips_and_rejects_oversize() { + use crate::wire::{encode_frame_header, parse_frame_header}; + let mut out = [0u8; 8]; + assert!(encode_frame_header(spec::wire::MSG_VIDEO_SLOT, 1, 1040, &mut out)); + let h = parse_frame_header(&out).expect("round trip"); + assert_eq!((h.kind, h.flags, h.len), (spec::wire::MSG_VIDEO_SLOT, 1, 1040)); + assert!(parse_frame_header(&out[..7]).is_none(), "short header"); + assert!( + !encode_frame_header(0x10, 0, spec::wire::MAX_PAYLOAD as u32 + 1, &mut out), + "oversize refused at encode" + ); + out[4..8].copy_from_slice(&(spec::wire::MAX_PAYLOAD as u32 + 1).to_le_bytes()); + assert!(parse_frame_header(&out).is_none(), "oversize refused at parse"); +} + +#[test] +fn wire_hello_and_beacon_parse() { + use crate::wire::{encode_hello, parse_beacon, parse_hello_ack}; + let mut out = [0u8; 80]; + let n = encode_hello("youtube", &mut out).expect("encodes"); + assert_eq!(n, 7 + 7); + assert_eq!(&out[0..4], &spec::wire::MAGIC.to_le_bytes()); + assert_eq!(out[4], spec::wire::VERSION); + assert_eq!(out[6] as usize, 7); + assert_eq!(&out[7..14], b"youtube"); + assert!(encode_hello("", &mut out).is_none()); + + let mut ack = [0u8; 8]; + ack[0..4].copy_from_slice(&spec::wire::MAGIC.to_le_bytes()); + ack[4] = 1; + assert_eq!(parse_hello_ack(&ack), Some(1)); + ack[0] = 0; + assert_eq!(parse_hello_ack(&ack), None); + + let mut beacon = Vec::new(); + beacon.extend_from_slice(&spec::wire::BEACON_MAGIC.to_le_bytes()); + beacon.push(spec::wire::VERSION); + beacon.push(0); + beacon.extend_from_slice(&spec::wire::PORT.to_le_bytes()); + beacon.push(7); + beacon.extend_from_slice(b"youtube"); + beacon.push(3); + beacon.extend_from_slice(b"Mac"); + let (port, app, name) = parse_beacon(&beacon).expect("beacon parses"); + assert_eq!((port, app, name), (spec::wire::PORT, "youtube", "Mac")); + // Truncations must fail cleanly at every length. + for cut in 0..beacon.len() { + assert!(parse_beacon(&beacon[..cut]).is_none(), "cut at {cut}"); + } +} + +#[test] +fn wire_payload_parsers_validate_and_survive_truncation() { + use crate::wire::{ + parse_audio_chunk, parse_file, parse_stream_mark, parse_stream_open, parse_video_slot, + }; + + // file: pathLen · path · blob + let mut file = Vec::new(); + file.extend_from_slice(&9u16.to_le_bytes()); + file.extend_from_slice(b"thumbs/a0"); + file.extend_from_slice(&[1, 2, 3, 4]); + let (path, blob) = parse_file(&file).expect("file parses"); + assert_eq!((path, blob), ("thumbs/a0", &[1u8, 2, 3, 4][..])); + for cut in 0..11 { + assert!(parse_file(&file[..cut]).is_none(), "cut at {cut}"); + } + + // streamOpen requires exactly a 96-byte header block after the path. + let block = stream_header_block(0, 0, 32, 16, 2, 0, 64, 4, 0); + let mut open = Vec::new(); + open.extend_from_slice(&7u16.to_le_bytes()); + open.extend_from_slice(b"media/v"); + open.extend_from_slice(&block); + let (path, got) = parse_stream_open(&open).expect("streamOpen parses"); + assert_eq!(path, "media/v"); + assert_eq!(got, &block[..]); + assert!(parse_stream_open(&open[..open.len() - 1]).is_none(), "short block refused"); + + // videoSlot: header · palette · indices. + let mut slot = Vec::new(); + slot.extend_from_slice(&2u32.to_le_bytes()); // seq + slot.extend_from_slice(&7u32.to_le_bytes()); // frameIndex + slot.extend_from_slice(&32u16.to_le_bytes()); + slot.extend_from_slice(&16u16.to_le_bytes()); + slot.extend_from_slice(&0u16.to_le_bytes()); // flags + slot.extend_from_slice(&0u16.to_le_bytes()); + slot.extend_from_slice(&[9u8; 1024]); + slot.extend_from_slice(&[5u8; 32 * 16]); + let msg = parse_video_slot(&slot).expect("slot parses"); + assert_eq!((msg.seq, msg.frame_index, msg.w, msg.h, msg.rle), (2, 7, 32, 16, false)); + assert_eq!(msg.palette.len(), 1024); + assert_eq!(msg.indices.len(), 32 * 16); + assert!(parse_video_slot(&slot[..1039]).is_none(), "short slot refused"); + let mut zero_seq = slot.clone(); + zero_seq[0..4].copy_from_slice(&0u32.to_le_bytes()); + assert!(parse_video_slot(&zero_seq).is_none(), "seq 0 refused"); + + // audioChunk + streamMark. + let mut chunk = Vec::new(); + chunk.extend_from_slice(&1u32.to_le_bytes()); + chunk.extend_from_slice(&2048u32.to_le_bytes()); + chunk.extend_from_slice(&[0u8; 64 * 2 * 2]); + let msg = parse_audio_chunk(&chunk).expect("chunk parses"); + assert_eq!((msg.seq, msg.start_frame, msg.pcm.len()), (1, 2048, 64 * 2 * 2)); + assert!(parse_audio_chunk(&chunk[..7]).is_none()); + + let mut mark = Vec::new(); + mark.extend_from_slice(&3u32.to_le_bytes()); + mark.extend_from_slice(&spec::wire::MARK_FLAG_ENDED.to_le_bytes()); + mark.extend_from_slice(&0u16.to_le_bytes()); + let msg = parse_stream_mark(&mark).expect("mark parses"); + assert!(msg.ended); + assert_eq!(msg.epoch, 3); + assert!(parse_stream_mark(&mark[..7]).is_none()); +} + +/// Independent reference writer: build the .pkst image with direct byte +/// writes straight off the spec.ts layout comment — a different code path +/// from RamStream, so equality below is a real check, not a tautology. +fn reference_write_slot( + buf: &mut [u8], + h: &crate::stream::StreamHeaders, + seq: u32, + frame_index: u32, + palette: &[u8], + indices: &[u8], +) { + let off = (h.video_off + ((seq - 1) % h.video.slot_count) * h.video.slot_size) as usize; + buf[off..off + 4].copy_from_slice(&seq.to_le_bytes()); + buf[off + 4..off + 8].copy_from_slice(&frame_index.to_le_bytes()); + buf[off + 8..off + 10].copy_from_slice(&(h.video.w as u16).to_le_bytes()); + buf[off + 10..off + 12].copy_from_slice(&(h.video.h as u16).to_le_bytes()); + for b in &mut buf[off + 12..off + 32] { + *b = 0; + } + buf[off + 32..off + 32 + 1024].copy_from_slice(palette); + buf[off + 1056..off + 1056 + indices.len()].copy_from_slice(indices); + let v = spec::stream::VRING_OFF; + buf[v + 20..v + 24].copy_from_slice(&seq.to_le_bytes()); +} + +fn reference_write_chunk( + buf: &mut [u8], + h: &crate::stream::StreamHeaders, + seq: u32, + start_frame: u32, + pcm: &[u8], +) { + let size = crate::stream::chunk_size(h.audio.chunk_frames, h.audio.channels).unwrap(); + let off = (h.audio_off + ((seq - 1) % h.audio.chunk_count) * size) as usize; + buf[off..off + 4].copy_from_slice(&seq.to_le_bytes()); + buf[off + 4..off + 8].copy_from_slice(&start_frame.to_le_bytes()); + for b in &mut buf[off + 8..off + 16] { + *b = 0; + } + buf[off + 16..off + 16 + pcm.len()].copy_from_slice(pcm); + let a = spec::stream::ARING_OFF; + buf[a + 20..a + 24].copy_from_slice(&seq.to_le_bytes()); +} + +#[test] +fn ram_stream_image_equals_the_reference_file_writer() { + use crate::wire::{AudioChunkMsg, StreamMarkMsg, VideoSlotMsg}; + + let block = stream_header_block(0, 0, 32, 16, 2, 0, 64, 2, 0); + let h = crate::stream::parse_header_block(&block).unwrap(); + let chunk_size = crate::stream::chunk_size(h.audio.chunk_frames, h.audio.channels).unwrap(); + let total = (h.audio_off + h.audio.chunk_count * chunk_size) as usize; + + // Reference image: header block + direct writes (3 slots so seq 3 laps + // slot 1's position, 2 chunks) + an epoch/ended mark. + let mut reference = alloc::vec![0u8; total]; + reference[..96].copy_from_slice(&block); + let pal = |seed: u8| -> Vec { (0..1024).map(|i| (i as u8).wrapping_add(seed)).collect() }; + let px = |seed: u8| -> Vec { (0..32 * 16).map(|i| (i as u8).wrapping_mul(seed)).collect() }; + let pcm = |seed: u8| -> Vec { (0..64 * 2 * 2).map(|i| (i as u8) ^ seed).collect() }; + for seq in 1..=3u32 { + reference_write_slot(&mut reference, &h, seq, seq * 2, &pal(seq as u8), &px(seq as u8)); + } + for seq in 1..=2u32 { + reference_write_chunk(&mut reference, &h, seq, seq * 64, &pcm(seq as u8)); + } + reference[8..12].copy_from_slice(&5u32.to_le_bytes()); // epoch + reference[6..8].copy_from_slice(&spec::stream::FLAG_ENDED.to_le_bytes()); + + // RamStream fed the same content as WIRE messages. + let mut ram = crate::stream_rx::RamStream::open(&block).expect("opens"); + for seq in 1..=3u32 { + let palette = pal(seq as u8); + let indices = px(seq as u8); + assert!(ram.apply_slot(&VideoSlotMsg { + seq, + frame_index: seq * 2, + w: 32, + h: 16, + rle: false, + palette: &palette, + indices: &indices, + })); + } + for seq in 1..=2u32 { + let bytes = pcm(seq as u8); + assert!(ram.apply_chunk(&AudioChunkMsg { seq, start_frame: seq * 64, pcm: &bytes })); + } + ram.apply_mark(&StreamMarkMsg { epoch: 5, ended: true }); + + assert_eq!(ram.buf().len(), reference.len()); + assert_eq!(ram.buf(), &reference[..], "RAM ring == file ring, byte for byte"); + + // And the shared readers see the expected world. + let live = crate::stream::parse_header_block(ram.buf()).unwrap(); + assert_eq!((live.epoch, live.ended), (5, true)); + assert_eq!(live.video.latest_seq, 3); + assert_eq!(live.audio.latest_seq, 2); + let off = crate::stream::slot_offset(&live, 3).unwrap() as usize; + let sh = crate::stream::parse_slot_header(&ram.buf()[off..], &live.video).unwrap(); + assert_eq!((sh.seq, sh.frame_index), (3, 6)); +} + +#[test] +fn ram_stream_decodes_rle_slots_and_rejects_bad_geometry() { + use crate::wire::VideoSlotMsg; + let block = stream_header_block(0, 0, 32, 16, 2, 0, 64, 2, 0); + let mut ram = crate::stream_rx::RamStream::open(&block).unwrap(); + + let raw: Vec = (0..32 * 16).map(|i| if i < 300 { 7 } else { (i % 5) as u8 }).collect(); + let rle = packbits_encode(&raw); + assert!(rle.len() < raw.len(), "fixture should actually compress"); + let palette = alloc::vec![1u8; 1024]; + assert!(ram.apply_slot(&VideoSlotMsg { + seq: 1, + frame_index: 0, + w: 32, + h: 16, + rle: true, + palette: &palette, + indices: &rle, + })); + let h = crate::stream::parse_header_block(ram.buf()).unwrap(); + let off = crate::stream::slot_offset(&h, 1).unwrap() as usize; + assert_eq!(&ram.buf()[off + 1056..off + 1056 + raw.len()], &raw[..]); + + // Wrong plane, wrong palette size, wrong index count, truncated RLE. + assert!(!ram.apply_slot(&VideoSlotMsg { + seq: 2, frame_index: 1, w: 16, h: 16, rle: false, palette: &palette, indices: &raw, + })); + assert!(!ram.apply_slot(&VideoSlotMsg { + seq: 2, frame_index: 1, w: 32, h: 16, rle: false, palette: &palette[..512], indices: &raw, + })); + assert!(!ram.apply_slot(&VideoSlotMsg { + seq: 2, frame_index: 1, w: 32, h: 16, rle: false, palette: &palette, indices: &raw[..100], + })); + assert!(!ram.apply_slot(&VideoSlotMsg { + seq: 2, frame_index: 1, w: 32, h: 16, rle: true, palette: &palette, indices: &rle[..rle.len() / 2], + })); + // Failures must not publish a cursor. + let h = crate::stream::parse_header_block(ram.buf()).unwrap(); + assert_eq!(h.video.latest_seq, 1); +} + +/// Cross-language: rebuild the committed TS-written .pkst golden through +/// RamStream (slots/chunks re-fed as WIRE messages) and require the byte- +/// identical file image — the socket transport IS the file transport. +#[test] +fn ram_stream_reconstructs_the_committed_golden() { + use crate::wire::{AudioChunkMsg, StreamMarkMsg, VideoSlotMsg}; + let golden: &[u8] = include_bytes!("../../../tests/fixtures/youtube-golden.pkst"); + let h = crate::stream::parse_header_block(golden).expect("golden parses"); + + // Open from a PRISTINE header (cursors zeroed, epoch 0, flags 0) — the + // exact block a streamOpen would carry before any writes. + let mut pristine = golden[..96].to_vec(); + pristine[6..8].copy_from_slice(&0u16.to_le_bytes()); + pristine[8..12].copy_from_slice(&0u32.to_le_bytes()); + let v = spec::stream::VRING_OFF; + let a = spec::stream::ARING_OFF; + pristine[v + 20..v + 24].copy_from_slice(&0u32.to_le_bytes()); + pristine[a + 20..a + 24].copy_from_slice(&0u32.to_le_bytes()); + let mut ram = crate::stream_rx::RamStream::open(&pristine).expect("opens"); + + // Re-feed every RESIDENT slot/chunk in seq order (the ring holds the + // last slot_count/chunk_count writes; lapped history is gone by design). + let mut slots: Vec<(u32, usize)> = Vec::new(); + for idx in 0..h.video.slot_count { + let off = (h.video_off + idx * h.video.slot_size) as usize; + if let Some(sh) = crate::stream::parse_slot_header(&golden[off..], &h.video) { + slots.push((sh.seq, off)); + } + } + slots.sort(); + for (seq, off) in slots { + let sh = crate::stream::parse_slot_header(&golden[off..], &h.video).unwrap(); + let pal = &golden[off + 32..off + 32 + 1024]; + let px = &golden[off + 1056..off + 1056 + (h.video.w * h.video.h) as usize]; + assert!(ram.apply_slot(&VideoSlotMsg { + seq, + frame_index: sh.frame_index, + w: h.video.w, + h: h.video.h, + rle: false, + palette: pal, + indices: px, + })); + } + let chunk_size = crate::stream::chunk_size(h.audio.chunk_frames, h.audio.channels).unwrap(); + let pcm_bytes = (h.audio.chunk_frames * h.audio.channels * 2) as usize; + let mut chunks: Vec<(u32, usize)> = Vec::new(); + for idx in 0..h.audio.chunk_count { + let off = (h.audio_off + idx * chunk_size) as usize; + if let Some(ch) = crate::stream::parse_chunk_header(&golden[off..]) { + chunks.push((ch.seq, off)); + } + } + chunks.sort(); + for (seq, off) in chunks { + let ch = crate::stream::parse_chunk_header(&golden[off..]).unwrap(); + assert!(ram.apply_chunk(&AudioChunkMsg { + seq, + start_frame: ch.start_frame, + pcm: &golden[off + 16..off + 16 + pcm_bytes], + })); + } + ram.apply_mark(&StreamMarkMsg { epoch: h.epoch, ended: h.ended }); + // Restore the golden's final cursors exactly (a lapped ring's latest may + // exceed the highest resident seq; apply_* published the resident max). + let final_v = h.video.latest_seq; + let final_a = h.audio.latest_seq; + let buf_len = ram.buf().len(); + assert_eq!(buf_len, golden.len(), "same preallocated image size"); + let mut image = ram.buf().to_vec(); + image[v + 20..v + 24].copy_from_slice(&final_v.to_le_bytes()); + image[a + 20..a + 24].copy_from_slice(&final_a.to_le_bytes()); + assert_eq!(&image[..], golden, "socket-fed RAM ring == TS-written file, byte for byte"); +} diff --git a/engine/core/src/wire.rs b/engine/core/src/wire.rs new file mode 100644 index 00000000..ed66fe32 --- /dev/null +++ b/engine/core/src/wire.rs @@ -0,0 +1,190 @@ +//! SVC WIRE protocol (PKNT) parsing — the svc mailbox over a socket (byte +//! layout in spec.ts "SVC WIRE protocol"). Pure bounds-checked readers and +//! fixed-size encoders over borrowed byte slices, the stream.rs discipline: +//! the host owns every socket and buffer; this module only interprets bytes, +//! so it stays no_std and unit-testable on any target. Hostile lengths +//! return None, never panic. + +use crate::spec::wire; +use crate::{rd_u16, rd_u32}; + +/// One decoded frame header (HEADER_SIZE bytes). +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub struct FrameHeader { + pub kind: u8, + pub flags: u8, + pub len: u32, +} + +/// Parse a frame header. None on short input or a payload length over +/// MAX_PAYLOAD (the reader must close the connection — resync inside a byte +/// stream is impossible). +pub fn parse_frame_header(b: &[u8]) -> Option { + if b.len() < wire::HEADER_SIZE { + return None; + } + let len = rd_u32(b, 4)?; + if len as usize > wire::MAX_PAYLOAD { + return None; + } + Some(FrameHeader { kind: b[0], flags: b[1], len }) +} + +/// Encode a frame header into `out` (exactly HEADER_SIZE bytes). +pub fn encode_frame_header(kind: u8, flags: u8, len: u32, out: &mut [u8]) -> bool { + if out.len() < wire::HEADER_SIZE || len as usize > wire::MAX_PAYLOAD { + return false; + } + out[0] = kind; + out[1] = flags; + out[2] = 0; + out[3] = 0; + out[4..8].copy_from_slice(&len.to_le_bytes()); + true +} + +/// The device's opening handshake: magic · version · reserved · appLen · app. +/// Returns the encoded length, or None when `out` is too small / the app id +/// does not fit its u8 length. +pub fn encode_hello(app: &str, out: &mut [u8]) -> Option { + let app = app.as_bytes(); + if app.is_empty() || app.len() > 64 { + return None; + } + let total = 7 + app.len(); + if out.len() < total { + return None; + } + out[0..4].copy_from_slice(&wire::MAGIC.to_le_bytes()); + out[4] = wire::VERSION; + out[5] = 0; + out[6] = app.len() as u8; + out[7..7 + app.len()].copy_from_slice(app); + Some(total) +} + +/// The host's handshake ack: magic · acceptedVersion · flags · reserved. +/// Returns the accepted version. +pub fn parse_hello_ack(b: &[u8]) -> Option { + if b.len() < 8 || rd_u32(b, 0)? != wire::MAGIC { + return None; + } + Some(b[4]) +} + +/// A `file` payload: svc-relative path + whole IMG-entry bytes. +pub fn parse_file<'a>(payload: &'a [u8]) -> Option<(&'a str, &'a [u8])> { + let path_len = rd_u16(payload, 0)? as usize; + let blob_at = 2usize.checked_add(path_len)?; + if payload.len() < blob_at { + return None; + } + let path = core::str::from_utf8(&payload[2..blob_at]).ok()?; + Some((path, &payload[blob_at..])) +} + +/// A `streamOpen` payload: announced path + the verbatim 96-byte header block. +pub fn parse_stream_open<'a>(payload: &'a [u8]) -> Option<(&'a str, &'a [u8])> { + let (path, block) = parse_file(payload)?; + if block.len() != crate::spec::stream::HEADER_BLOCK_SIZE { + return None; + } + Some((path, block)) +} + +/// A `videoSlot` payload (leading fields + palette + indices). +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub struct VideoSlotMsg<'a> { + pub seq: u32, + pub frame_index: u32, + pub w: u32, + pub h: u32, + /// Indices are PackBits-RLE (decode to exactly w*h) instead of raw. + pub rle: bool, + pub palette: &'a [u8], + pub indices: &'a [u8], +} + +pub fn parse_video_slot<'a>(payload: &'a [u8]) -> Option> { + let palette_at = wire::SLOT_HEADER_SIZE; + let indices_at = palette_at.checked_add(1024)?; + if payload.len() < indices_at { + return None; + } + let seq = rd_u32(payload, 0)?; + let frame_index = rd_u32(payload, 4)?; + let w = rd_u16(payload, 8)? as u32; + let h = rd_u16(payload, 10)? as u32; + let flags = rd_u16(payload, 12)?; + if seq == 0 { + return None; + } + Some(VideoSlotMsg { + seq, + frame_index, + w, + h, + rle: flags & wire::SLOT_FLAG_RLE != 0, + palette: &payload[palette_at..indices_at], + indices: &payload[indices_at..], + }) +} + +/// An `audioChunk` payload (leading fields + interleaved s16 PCM bytes). +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub struct AudioChunkMsg<'a> { + pub seq: u32, + pub start_frame: u32, + pub pcm: &'a [u8], +} + +pub fn parse_audio_chunk<'a>(payload: &'a [u8]) -> Option> { + if payload.len() < wire::CHUNK_HEADER_SIZE { + return None; + } + let seq = rd_u32(payload, 0)?; + if seq == 0 { + return None; + } + Some(AudioChunkMsg { + seq, + start_frame: rd_u32(payload, 4)?, + pcm: &payload[wire::CHUNK_HEADER_SIZE..], + }) +} + +/// A `streamMark` payload: epoch bump / ended marker. +#[derive(Clone, Copy, PartialEq, Eq, Debug)] +pub struct StreamMarkMsg { + pub epoch: u32, + pub ended: bool, +} + +pub fn parse_stream_mark(payload: &[u8]) -> Option { + if payload.len() < wire::MARK_SIZE { + return None; + } + Some(StreamMarkMsg { + epoch: rd_u32(payload, 0)?, + ended: rd_u16(payload, 4)? & wire::MARK_FLAG_ENDED != 0, + }) +} + +/// A discovery beacon datagram. Returns (tcpPort, app, displayName). +pub fn parse_beacon<'a>(datagram: &'a [u8]) -> Option<(u16, &'a str, &'a str)> { + if rd_u32(datagram, 0)? != wire::BEACON_MAGIC || datagram.get(4).copied()? != wire::VERSION { + return None; + } + let port = rd_u16(datagram, 6)?; + let app_len = datagram.get(8).copied()? as usize; + let name_len_at = 9usize.checked_add(app_len)?; + let name_len = datagram.get(name_len_at).copied()? as usize; + let name_at = name_len_at + 1; + let end = name_at.checked_add(name_len)?; + if datagram.len() < end { + return None; + } + let app = core::str::from_utf8(&datagram[9..name_len_at]).ok()?; + let name = core::str::from_utf8(&datagram[name_at..end]).ok()?; + Some((port, app, name)) +}