From d58d0bb0978f3627cf7ae07bb2cfcbfb276dcb94 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 19 May 2026 09:07:29 +0100 Subject: [PATCH] Expose rolling reservoir view reads --- docs/perfect-machine.md | 1 + native/README.md | 5 +- .../reservoir/include/localcast_reservoir.h | 14 ++ native/reservoir/src/lib.rs | 174 ++++++++++++++++++ state/evidence.jsonl | 1 + tests/test_perfect_machine_contract.py | 2 + 6 files changed, 195 insertions(+), 2 deletions(-) diff --git a/docs/perfect-machine.md b/docs/perfect-machine.md index 6895012..8319b8d 100644 --- a/docs/perfect-machine.md +++ b/docs/perfect-machine.md @@ -174,6 +174,7 @@ in `native/reservoir/include/localcast_reservoir.h`: - push timestamped sample metadata into a typed view of the rolling buffer; - advance/query the shared edge and window start; - query total length and typed-view length; +- read samples by total rolling-buffer index or typed-view index; - query the latest sample for a sensor hash. - create/destroy an opaque `LocalcastRuntime`; - append typed camera/audio/phase/event/render handles through producer calls; diff --git a/native/README.md b/native/README.md index a73fa55..dd6e48f 100644 --- a/native/README.md +++ b/native/README.md @@ -24,8 +24,9 @@ The reservoir ABI is intentionally small: camera frame, camera feature, scene ray, surface claim, material claim, audio block, phase claim, event claim, and render packet. - The reservoir can create/destroy, push, set/query edge, query window start, - count the whole rolling buffer, count a typed view, and fetch the latest - sample for a sensor hash inside a typed view. + count the whole rolling buffer, count a typed view, read samples by total + buffer index or typed-view index, and fetch the latest sample for a sensor + hash inside a typed view. ## Runtime ABI diff --git a/native/reservoir/include/localcast_reservoir.h b/native/reservoir/include/localcast_reservoir.h index f73fe0e..5fc430f 100644 --- a/native/reservoir/include/localcast_reservoir.h +++ b/native/reservoir/include/localcast_reservoir.h @@ -61,6 +61,20 @@ uint64_t localcast_reservoir_window_start_ns(const LocalcastReservoir *reservoir size_t localcast_reservoir_len(const LocalcastReservoir *reservoir); size_t localcast_reservoir_view_len(const LocalcastReservoir *reservoir, uint32_t sample_kind); +bool localcast_reservoir_sample_at( + const LocalcastReservoir *reservoir, + size_t index, + uint32_t *out_sample_kind, + LocalcastSampleHandle *out_sample +); + +bool localcast_reservoir_view_sample_at( + const LocalcastReservoir *reservoir, + uint32_t sample_kind, + size_t index, + LocalcastSampleHandle *out_sample +); + bool localcast_reservoir_latest_for_sensor( const LocalcastReservoir *reservoir, uint32_t sample_kind, diff --git a/native/reservoir/src/lib.rs b/native/reservoir/src/lib.rs index 218aaf5..b8b97cb 100644 --- a/native/reservoir/src/lib.rs +++ b/native/reservoir/src/lib.rs @@ -113,6 +113,10 @@ impl RollingReservoir { self.samples.iter() } + pub fn sample_at(&self, index: usize) -> Option<&ReservoirSample> { + self.samples.get(index) + } + pub fn view(&self, kind: SampleKind) -> impl Iterator> { self.samples .iter() @@ -123,6 +127,10 @@ impl RollingReservoir { self.view(kind).count() } + pub fn view_sample_at(&self, kind: SampleKind, index: usize) -> Option<&ReservoirSample> { + self.view(kind).nth(index) + } + pub fn latest_for_sensor( &self, kind: SampleKind, @@ -329,6 +337,53 @@ pub extern "C" fn localcast_reservoir_view_len( reservoir.inner.view_len(kind) } +#[unsafe(no_mangle)] +pub extern "C" fn localcast_reservoir_sample_at( + ptr: *const LocalcastReservoir, + index: usize, + out_sample_kind: *mut u32, + out_sample: *mut LocalcastSampleHandle, +) -> bool { + let Some(reservoir) = (unsafe { ptr.as_ref() }) else { + return false; + }; + let Some(out_kind) = (unsafe { out_sample_kind.as_mut() }) else { + return false; + }; + let Some(out) = (unsafe { out_sample.as_mut() }) else { + return false; + }; + let Some(sample) = reservoir.inner.sample_at(index) else { + return false; + }; + *out_kind = sample.kind as u32; + *out = sample.payload; + true +} + +#[unsafe(no_mangle)] +pub extern "C" fn localcast_reservoir_view_sample_at( + ptr: *const LocalcastReservoir, + sample_kind: u32, + index: usize, + out_sample: *mut LocalcastSampleHandle, +) -> bool { + let Some(reservoir) = (unsafe { ptr.as_ref() }) else { + return false; + }; + let Some(kind) = SampleKind::from_u32(sample_kind) else { + return false; + }; + let Some(out) = (unsafe { out_sample.as_mut() }) else { + return false; + }; + let Some(sample) = reservoir.inner.view_sample_at(kind, index) else { + return false; + }; + *out = sample.payload; + true +} + #[unsafe(no_mangle)] pub extern "C" fn localcast_reservoir_latest_for_sensor( ptr: *const LocalcastReservoir, @@ -539,6 +594,49 @@ mod tests { assert_eq!(3, reservoir.len()); } + #[test] + fn indexed_reads_expose_total_buffer_and_typed_views() { + let mut reservoir = RollingReservoir::new(DEFAULT_RESERVOIR_NS); + reservoir.push(ReservoirSample::new( + SampleKind::CameraFrame, + "cam-a", + 10, + 11, + 0, + "frame-a", + )); + reservoir.push(ReservoirSample::new( + SampleKind::AudioBlock, + "mic", + 12, + 13, + 0, + "audio", + )); + reservoir.push(ReservoirSample::new( + SampleKind::CameraFrame, + "cam-b", + 14, + 15, + 0, + "frame-b", + )); + + assert_eq!("audio", reservoir.sample_at(1).unwrap().payload); + assert_eq!( + "frame-b", + reservoir + .view_sample_at(SampleKind::CameraFrame, 1) + .unwrap() + .payload + ); + assert!( + reservoir + .view_sample_at(SampleKind::PhaseClaim, 0) + .is_none() + ); + } + #[test] fn latest_for_sensor_is_scoped_to_kind_and_sensor() { let mut reservoir = RollingReservoir::new(DEFAULT_RESERVOIR_NS); @@ -635,6 +733,82 @@ mod tests { localcast_reservoir_destroy(reservoir); } + #[test] + fn c_abi_reads_total_buffer_and_view_samples_by_index() { + let reservoir = localcast_reservoir_create(DEFAULT_RESERVOIR_NS); + assert!(localcast_reservoir_push( + reservoir, + 0, + LocalcastSampleHandle { + sensor_id_hash: 10, + timestamp_ns: 10, + arrival_ns: 11, + sequence: 0, + payload_handle: 100, + }, + )); + assert!(localcast_reservoir_push( + reservoir, + 5, + LocalcastSampleHandle { + sensor_id_hash: 20, + timestamp_ns: 12, + arrival_ns: 13, + sequence: 0, + payload_handle: 200, + }, + )); + assert!(localcast_reservoir_push( + reservoir, + 0, + LocalcastSampleHandle { + sensor_id_hash: 30, + timestamp_ns: 14, + arrival_ns: 15, + sequence: 0, + payload_handle: 300, + }, + )); + + let mut kind = 99; + let mut sample = LocalcastSampleHandle { + sensor_id_hash: 0, + timestamp_ns: 0, + arrival_ns: 0, + sequence: 0, + payload_handle: 0, + }; + + assert!(localcast_reservoir_sample_at( + reservoir, + 1, + &mut kind, + &mut sample, + )); + assert_eq!(5, kind); + assert_eq!(200, sample.payload_handle); + assert!(localcast_reservoir_view_sample_at( + reservoir, + 0, + 1, + &mut sample, + )); + assert_eq!(300, sample.payload_handle); + assert!(!localcast_reservoir_sample_at( + reservoir, + 99, + &mut kind, + &mut sample, + )); + assert!(!localcast_reservoir_view_sample_at( + reservoir, + 99, + 0, + &mut sample, + )); + localcast_reservoir_destroy(reservoir); + } + #[test] fn c_abi_rejects_nulls_and_unknown_sample_kinds() { let sample = LocalcastSampleHandle { diff --git a/state/evidence.jsonl b/state/evidence.jsonl index d34d2dc..51b2e83 100644 --- a/state/evidence.jsonl +++ b/state/evidence.jsonl @@ -56,3 +56,4 @@ {"ts": "2026-05-19T07:57:59+00:00", "type": "implementation", "status": "ok", "note": "OpenGL Spout sink quarantined: localcast.sensor_fusion no longer exports spout_output surfaces; the old SpoutGL renderer now lives under localcast.diagnostics.spout_output and scripts/stream_spout imports it as diagnostic/migration code only."} {"ts": "2026-05-19T08:01:04+00:00", "type": "implementation", "status": "ok", "note": "Python visual producer quarantined by name: scripts/live_sensor_fusion.py is now scripts/diagnostic_live_sensor_fusion.py, tests/docs import the diagnostic path, and future native/Aquarium work should delete or split its pure pieces rather than extend it."} {"ts": "2026-05-19T08:03:24+00:00", "type": "implementation", "status": "ok", "note": "Diagnostic Spout launcher renamed: scripts/stream_spout.py is now scripts/diagnostic_stream_spout.py, keeping the OpenGL bridge visibly outside the production path while Aquarium Spout publication remains the target."} +{"ts": "2026-05-19T08:07:19+00:00", "type": "implementation", "status": "ok", "note": "Native reservoir ABI now exposes indexed reads over the total rolling buffer and typed views via localcast_reservoir_sample_at and localcast_reservoir_view_sample_at, allowing Aquarium/Faust consumers to walk the live window without owning private histories."} diff --git a/tests/test_perfect_machine_contract.py b/tests/test_perfect_machine_contract.py index 60d43c9..40acf8c 100644 --- a/tests/test_perfect_machine_contract.py +++ b/tests/test_perfect_machine_contract.py @@ -45,6 +45,8 @@ class PerfectMachineContractTests(unittest.TestCase): self.assertIn("enum LocalcastSampleKind", header) self.assertIn("size_t localcast_reservoir_len", header) self.assertIn("size_t localcast_reservoir_view_len", header) + self.assertIn("bool localcast_reservoir_sample_at", header) + self.assertIn("bool localcast_reservoir_view_sample_at", header) self.assertIn("bool localcast_reservoir_latest_for_sensor", header) def test_native_runtime_header_exposes_typed_producer_spine(self):