agentos_v8_runtime/
runtime_protocol.rs

1use crate::ipc_binary::{BinaryFrame, ExecutionErrorBin};
2use std::io;
3use std::sync::Arc;
4
5#[derive(Debug, Clone, PartialEq, Eq)]
6pub struct WarmSessionHint {
7    pub bridge_code: String,
8    pub userland_code: String,
9    pub heap_limit_mb: Option<u32>,
10}
11
12#[derive(Debug, Clone, PartialEq)]
13pub enum RuntimeCommand {
14    CreateSession {
15        session_id: String,
16        heap_limit_mb: Option<u32>,
17        cpu_time_limit_ms: Option<u32>,
18        wall_clock_limit_ms: Option<u32>,
19        warm_hint: Option<WarmSessionHint>,
20    },
21    DestroySession {
22        session_id: String,
23    },
24    WarmSnapshot {
25        bridge_code: String,
26        userland_code: String,
27    },
28    SendToSession {
29        session_id: String,
30        message: SessionMessage,
31    },
32    /// Install a direct module-source reader on a session thread. Carries the live
33    /// reader (not serialized) so module loads skip the bridge round-trip. Routed
34    /// through the dispatch thread (which owns the session manager) like
35    /// `SendToSession`, then forwarded as a `SessionCommand::SetModuleReader`.
36    SetSessionModuleReader {
37        session_id: String,
38        reader: ModuleReaderHandle,
39    },
40}
41
42/// Wrapper that lets a live `GuestModuleReader` ride `RuntimeCommand` despite its
43/// `derive(Debug, Clone, PartialEq)`: the trait object lives behind an
44/// `Arc<Mutex<Option<..>>>`, and the dispatch thread `take()`s it out exactly once.
45#[derive(Clone)]
46pub struct ModuleReaderHandle(
47    std::sync::Arc<std::sync::Mutex<Option<Box<dyn crate::execution::GuestModuleReader>>>>,
48);
49
50impl ModuleReaderHandle {
51    pub fn new(reader: Box<dyn crate::execution::GuestModuleReader>) -> Self {
52        ModuleReaderHandle(std::sync::Arc::new(std::sync::Mutex::new(Some(reader))))
53    }
54
55    /// Take the reader out (dispatch-thread side); `None` if already taken.
56    pub fn take(&self) -> Option<Box<dyn crate::execution::GuestModuleReader>> {
57        self.0.lock().ok().and_then(|mut guard| guard.take())
58    }
59}
60
61impl std::fmt::Debug for ModuleReaderHandle {
62    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
63        f.write_str("ModuleReaderHandle(..)")
64    }
65}
66
67impl PartialEq for ModuleReaderHandle {
68    fn eq(&self, other: &Self) -> bool {
69        std::sync::Arc::ptr_eq(&self.0, &other.0)
70    }
71}
72
73#[derive(Debug, Clone, PartialEq)]
74pub enum SessionMessage {
75    InjectGlobals {
76        payload: Vec<u8>,
77    },
78    Execute {
79        mode: u8,
80        file_path: String,
81        bridge_code: String,
82        post_restore_script: String,
83        userland_code: String,
84        high_resolution_time: bool,
85        user_code: String,
86        wasm_module_bytes: Option<Arc<Vec<u8>>>,
87    },
88    BridgeResponse(BridgeResponse),
89    StreamEvent(StreamEvent),
90    TerminateExecution,
91}
92
93#[derive(Debug, Clone, PartialEq)]
94pub struct BridgeResponse {
95    pub call_id: u64,
96    pub status: u8,
97    pub payload: Vec<u8>,
98}
99
100#[derive(Debug, Clone, PartialEq)]
101pub struct StreamEvent {
102    pub event_type: String,
103    pub payload: Vec<u8>,
104}
105
106#[derive(Debug, Clone, PartialEq)]
107pub enum RuntimeEvent {
108    BridgeCall {
109        session_id: String,
110        call_id: u64,
111        method: String,
112        payload: Vec<u8>,
113    },
114    ExecutionResult {
115        session_id: String,
116        exit_code: i32,
117        exports: Option<Vec<u8>>,
118        error: Option<ExecutionErrorBin>,
119    },
120    Log {
121        session_id: String,
122        channel: u8,
123        message: String,
124    },
125    StreamCallback {
126        session_id: String,
127        callback_type: String,
128        payload: Vec<u8>,
129    },
130}
131
132impl RuntimeEvent {
133    pub fn session_id(&self) -> &str {
134        match self {
135            RuntimeEvent::BridgeCall { session_id, .. }
136            | RuntimeEvent::ExecutionResult { session_id, .. }
137            | RuntimeEvent::Log { session_id, .. }
138            | RuntimeEvent::StreamCallback { session_id, .. } => session_id,
139        }
140    }
141}
142
143impl TryFrom<BinaryFrame> for RuntimeCommand {
144    type Error = io::Error;
145
146    fn try_from(frame: BinaryFrame) -> Result<Self, Self::Error> {
147        match frame {
148            BinaryFrame::CreateSession {
149                session_id,
150                heap_limit_mb,
151                cpu_time_limit_ms,
152                wall_clock_limit_ms,
153            } => Ok(RuntimeCommand::CreateSession {
154                session_id,
155                heap_limit_mb: non_zero_option(heap_limit_mb),
156                cpu_time_limit_ms: non_zero_option(cpu_time_limit_ms),
157                wall_clock_limit_ms: non_zero_option(wall_clock_limit_ms),
158                warm_hint: None,
159            }),
160            BinaryFrame::DestroySession { session_id } => {
161                Ok(RuntimeCommand::DestroySession { session_id })
162            }
163            BinaryFrame::InjectGlobals {
164                session_id,
165                payload,
166            } => Ok(RuntimeCommand::SendToSession {
167                session_id,
168                message: SessionMessage::InjectGlobals { payload },
169            }),
170            BinaryFrame::Execute {
171                session_id,
172                mode,
173                file_path,
174                bridge_code,
175                post_restore_script,
176                userland_code,
177                high_resolution_time,
178                user_code,
179            } => {
180                if mode > 1 {
181                    return Err(io::Error::new(
182                        io::ErrorKind::InvalidInput,
183                        format!("unknown Execute mode: {mode}"),
184                    ));
185                }
186                Ok(RuntimeCommand::SendToSession {
187                    session_id,
188                    message: SessionMessage::Execute {
189                        mode,
190                        file_path,
191                        bridge_code,
192                        post_restore_script,
193                        userland_code,
194                        high_resolution_time,
195                        user_code,
196                        wasm_module_bytes: None,
197                    },
198                })
199            }
200            BinaryFrame::BridgeResponse {
201                session_id,
202                call_id,
203                status,
204                payload,
205            } => {
206                validate_bridge_response_status(status)?;
207                Ok(RuntimeCommand::SendToSession {
208                    session_id,
209                    message: SessionMessage::BridgeResponse(BridgeResponse {
210                        call_id,
211                        status,
212                        payload,
213                    }),
214                })
215            }
216            BinaryFrame::StreamEvent {
217                session_id,
218                event_type,
219                payload,
220            } => Ok(RuntimeCommand::SendToSession {
221                session_id,
222                message: SessionMessage::StreamEvent(StreamEvent {
223                    event_type,
224                    payload,
225                }),
226            }),
227            BinaryFrame::TerminateExecution { session_id } => Ok(RuntimeCommand::SendToSession {
228                session_id,
229                message: SessionMessage::TerminateExecution,
230            }),
231            BinaryFrame::WarmSnapshot {
232                bridge_code,
233                userland_code,
234            } => Ok(RuntimeCommand::WarmSnapshot {
235                bridge_code,
236                userland_code,
237            }),
238            BinaryFrame::Authenticate { .. } => Err(io::Error::new(
239                io::ErrorKind::InvalidInput,
240                "Authenticate is not supported by the embedded runtime",
241            )),
242            _ => Err(io::Error::new(
243                io::ErrorKind::InvalidInput,
244                "host-output frames cannot be sent into the embedded runtime",
245            )),
246        }
247    }
248}
249
250impl From<RuntimeEvent> for BinaryFrame {
251    fn from(event: RuntimeEvent) -> Self {
252        match event {
253            RuntimeEvent::BridgeCall {
254                session_id,
255                call_id,
256                method,
257                payload,
258            } => BinaryFrame::BridgeCall {
259                session_id,
260                call_id,
261                method,
262                payload,
263            },
264            RuntimeEvent::ExecutionResult {
265                session_id,
266                exit_code,
267                exports,
268                error,
269            } => BinaryFrame::ExecutionResult {
270                session_id,
271                exit_code,
272                exports,
273                error,
274            },
275            RuntimeEvent::Log {
276                session_id,
277                channel,
278                message,
279            } => BinaryFrame::Log {
280                session_id,
281                channel,
282                message,
283            },
284            RuntimeEvent::StreamCallback {
285                session_id,
286                callback_type,
287                payload,
288            } => BinaryFrame::StreamCallback {
289                session_id,
290                callback_type,
291                payload,
292            },
293        }
294    }
295}
296
297fn non_zero_option(value: u32) -> Option<u32> {
298    if value == 0 {
299        None
300    } else {
301        Some(value)
302    }
303}
304
305pub fn validate_bridge_response_status(status: u8) -> io::Result<()> {
306    if status <= 2 {
307        return Ok(());
308    }
309    Err(io::Error::new(
310        io::ErrorKind::InvalidInput,
311        format!("unknown BridgeResponse status: {status}"),
312    ))
313}
314
315#[cfg(test)]
316mod tests {
317    use super::*;
318
319    #[test]
320    fn rejects_unknown_execute_mode() {
321        let err = RuntimeCommand::try_from(BinaryFrame::Execute {
322            session_id: "s".into(),
323            mode: 2,
324            file_path: "/app/main.mjs".into(),
325            bridge_code: String::new(),
326            post_restore_script: String::new(),
327            userland_code: String::new(),
328            high_resolution_time: false,
329            user_code: String::new(),
330        })
331        .expect_err("unknown execute mode should be rejected");
332
333        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
334        assert!(err.to_string().contains("unknown Execute mode"));
335    }
336
337    #[test]
338    fn rejects_unknown_bridge_response_status() {
339        let err = RuntimeCommand::try_from(BinaryFrame::BridgeResponse {
340            session_id: "s".into(),
341            call_id: 1,
342            status: 3,
343            payload: Vec::new(),
344        })
345        .expect_err("unknown bridge response status should be rejected");
346
347        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
348        assert!(err.to_string().contains("unknown BridgeResponse status"));
349    }
350
351    #[test]
352    fn accepts_known_bridge_response_statuses() {
353        for status in 0..=2 {
354            let command = RuntimeCommand::try_from(BinaryFrame::BridgeResponse {
355                session_id: "s".into(),
356                call_id: 1,
357                status,
358                payload: Vec::new(),
359            })
360            .expect("known bridge response status should be accepted");
361
362            assert!(matches!(
363                command,
364                RuntimeCommand::SendToSession {
365                    message: SessionMessage::BridgeResponse(BridgeResponse { status: s, .. }),
366                    ..
367                } if s == status
368            ));
369        }
370    }
371}
agentos_v8_runtime/runtime_protocol.rs

agentos_v8_runtime/
runtime_protocol.rs
