agentos_v8_runtime/
runtime_protocol.rs

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

agentos_v8_runtime/
runtime_protocol.rs
