implements PLN-0113

crates/console/prometeu-system/src/lib.rs

@@ -9,5 +9,7 @@ pub use programs::{NativeShellApp, PrometeuHub, SystemProfileAction, SystemProfi
 pub use services::fs;
 pub use services::process;
 pub use services::task;
-pub use services::vm_runtime::VirtualMachineRuntime;
+pub use services::vm_runtime::{
+    RenderWorkerHandoff, RenderWorkerHandoffWait, VirtualMachineRuntime,
+};
 pub use services::windows;

crates/console/prometeu-system/src/services/vm_runtime/mod.rs

@@ -4,6 +4,7 @@ mod dispatch;
 mod frame_scheduler;
 mod lifecycle;
 pub mod render_manager;
+pub mod render_worker_handoff;
 #[cfg(test)]
 mod tests;
 mod tick;
@@ -17,6 +18,7 @@ use prometeu_hal::telemetry::{AtomicTelemetry, CertificationConfig, Certifier};
 use prometeu_hal::{Gfx2dCommand, GfxUiCommand};
 use prometeu_vm::VirtualMachine;
 pub use render_manager::{RenderManager, RenderRuntimeCapabilities};
+pub use render_worker_handoff::{RenderWorkerHandoff, RenderWorkerHandoffWait};
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::sync::atomic::AtomicU32;

crates/console/prometeu-system/src/services/vm_runtime/render_worker_handoff.rs

@@ -0,0 +1,172 @@
+use prometeu_hal::{RenderSubmission, RenderWorkerTelemetry};
+use std::sync::{Condvar, Mutex};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum RenderWorkerHandoffWait {
+    Submission(RenderSubmission),
+    Shutdown,
+}
+
+#[derive(Debug, Default)]
+struct RenderWorkerHandoffState {
+    pending: Option<RenderSubmission>,
+    shutdown_requested: bool,
+    telemetry: RenderWorkerTelemetry,
+}
+
+#[derive(Debug, Default)]
+pub struct RenderWorkerHandoff {
+    state: Mutex<RenderWorkerHandoffState>,
+    ready: Condvar,
+}
+
+impl RenderWorkerHandoff {
+    pub fn publish(&self, submission: RenderSubmission) {
+        let mut state = self.state.lock().unwrap();
+        if let Some(replaced) = state.pending.replace(submission) {
+            state.telemetry.record_replaced_before_consume(replaced.frame_id.get());
+        }
+        let frame_id = state.pending.as_ref().expect("submission was just stored").frame_id;
+        state.telemetry.record_produced(frame_id.get());
+        self.ready.notify_one();
+    }
+
+    pub fn take_latest(&self) -> Option<RenderSubmission> {
+        let mut state = self.state.lock().unwrap();
+        let submission = state.pending.take()?;
+        state.telemetry.record_consumed(submission.frame_id.get());
+        Some(submission)
+    }
+
+    pub fn wait_take(&self) -> RenderWorkerHandoffWait {
+        self.wait_take_with_hook(|| {})
+    }
+
+    pub fn request_shutdown(&self) {
+        let mut state = self.state.lock().unwrap();
+        state.shutdown_requested = true;
+        self.ready.notify_all();
+    }
+
+    pub fn telemetry(&self) -> RenderWorkerTelemetry {
+        self.state.lock().unwrap().telemetry
+    }
+
+    fn wait_take_with_hook(&self, mut before_wait: impl FnMut()) -> RenderWorkerHandoffWait {
+        let mut state = self.state.lock().unwrap();
+        loop {
+            if let Some(submission) = state.pending.take() {
+                state.telemetry.record_consumed(submission.frame_id.get());
+                return RenderWorkerHandoffWait::Submission(submission);
+            }
+            if state.shutdown_requested {
+                return RenderWorkerHandoffWait::Shutdown;
+            }
+            before_wait();
+            state = self.ready.wait(state).unwrap();
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use prometeu_hal::{FrameId, Game2DFramePacket};
+    use std::sync::Arc;
+    use std::sync::mpsc;
+    use std::thread;
+
+    fn submission(frame_id: u64) -> RenderSubmission {
+        RenderSubmission::game2d(FrameId::new(frame_id), Game2DFramePacket::default())
+    }
+
+    #[test]
+    fn render_worker_handoff_takes_owned_submission() {
+        let handoff = RenderWorkerHandoff::default();
+
+        assert!(handoff.take_latest().is_none());
+        handoff.publish(submission(7));
+
+        let taken = handoff.take_latest().expect("pending submission");
+        assert_eq!(taken.frame_id, FrameId::new(7));
+        assert!(handoff.take_latest().is_none());
+        assert_eq!(handoff.telemetry().produced_submissions, 1);
+        assert_eq!(handoff.telemetry().consumed_submissions, 1);
+    }
+
+    #[test]
+    fn render_worker_handoff_wait_take_returns_pending_without_blocking() {
+        let handoff = RenderWorkerHandoff::default();
+        handoff.publish(submission(8));
+
+        assert_eq!(handoff.wait_take(), RenderWorkerHandoffWait::Submission(submission(8)));
+        assert!(handoff.take_latest().is_none());
+    }
+
+    #[test]
+    fn render_worker_handoff_is_latest_wins_and_counts_replacements() {
+        let handoff = RenderWorkerHandoff::default();
+
+        handoff.publish(submission(1));
+        handoff.publish(submission(2));
+        handoff.publish(submission(3));
+
+        let taken = handoff.take_latest().expect("latest submission");
+        let telemetry = handoff.telemetry();
+        assert_eq!(taken.frame_id, FrameId::new(3));
+        assert_eq!(telemetry.produced_submissions, 3);
+        assert_eq!(telemetry.replaced_before_consume, 2);
+        assert_eq!(telemetry.consumed_submissions, 1);
+        assert_eq!(telemetry.last_produced_frame_id, 3);
+        assert_eq!(telemetry.last_dropped_frame_id, 2);
+    }
+
+    #[test]
+    fn render_worker_handoff_waits_for_publish_without_sleep() {
+        let handoff = Arc::new(RenderWorkerHandoff::default());
+        let consumer_handoff = Arc::clone(&handoff);
+        let (waiting_tx, waiting_rx) = mpsc::channel();
+
+        let consumer = thread::spawn(move || {
+            let mut waiting_tx = Some(waiting_tx);
+            consumer_handoff.wait_take_with_hook(|| {
+                if let Some(tx) = waiting_tx.take() {
+                    tx.send(()).unwrap();
+                }
+            })
+        });
+
+        waiting_rx.recv().expect("consumer reached condvar wait");
+        handoff.publish(submission(42));
+
+        assert_eq!(
+            consumer.join().expect("consumer thread should finish"),
+            RenderWorkerHandoffWait::Submission(submission(42))
+        );
+        assert_eq!(handoff.telemetry().consumed_submissions, 1);
+    }
+
+    #[test]
+    fn render_worker_handoff_wait_returns_shutdown_signal() {
+        let handoff = Arc::new(RenderWorkerHandoff::default());
+        let consumer_handoff = Arc::clone(&handoff);
+        let (waiting_tx, waiting_rx) = mpsc::channel();
+
+        let consumer = thread::spawn(move || {
+            let mut waiting_tx = Some(waiting_tx);
+            consumer_handoff.wait_take_with_hook(|| {
+                if let Some(tx) = waiting_tx.take() {
+                    tx.send(()).unwrap();
+                }
+            })
+        });
+
+        waiting_rx.recv().expect("consumer reached condvar wait");
+        handoff.request_shutdown();
+
+        assert_eq!(
+            consumer.join().expect("consumer thread should finish"),
+            RenderWorkerHandoffWait::Shutdown
+        );
+    }
+}

discussion/workflow/plans/PLN-0113-implement-thread-safe-latest-wins-handoff.md

@@ -2,7 +2,7 @@
 id: PLN-0113
 ticket: real-render-worker-establishment
 title: Implement Thread-Safe Latest-Wins Handoff
-status: open
+status: done
 created: 2026-06-15
 completed:
 ref_decisions: [DEC-0033]