agentos_v8_runtime/
runtime_protocol.rs

1use crate::ipc_binary::{BinaryFrame, ExecutionErrorBin};
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    },
151    Log {
152        session_id: String,
153        channel: u8,
154        message: String,
155    },
156    StreamCallback {
157        session_id: String,
158        callback_type: String,
159        payload: Vec<u8>,
160    },
161}
162
163impl RuntimeEvent {
164    pub fn session_id(&self) -> &str {
165        match self {
166            RuntimeEvent::BridgeCall { session_id, .. }
167            | RuntimeEvent::ExecutionResult { session_id, .. }
168            | RuntimeEvent::Log { session_id, .. }
169            | RuntimeEvent::StreamCallback { session_id, .. } => session_id,
170        }
171    }
172}
173
174impl TryFrom<BinaryFrame> for RuntimeCommand {
175    type Error = io::Error;
176
177    fn try_from(frame: BinaryFrame) -> Result<Self, Self::Error> {
178        match frame {
179            BinaryFrame::CreateSession {
180                session_id,
181                heap_limit_mb,
182                cpu_time_limit_ms,
183                wall_clock_limit_ms,
184            } => Ok(RuntimeCommand::CreateSession {
185                session_id,
186                heap_limit_mb: non_zero_option(heap_limit_mb),
187                cpu_time_limit_ms: non_zero_option(cpu_time_limit_ms),
188                wall_clock_limit_ms: non_zero_option(wall_clock_limit_ms),
189                warm_hint: None,
190            }),
191            BinaryFrame::DestroySession { session_id } => {
192                Ok(RuntimeCommand::DestroySession { session_id })
193            }
194            BinaryFrame::InjectGlobals {
195                session_id,
196                payload,
197            } => Ok(RuntimeCommand::SendToSession {
198                session_id,
199                message: SessionMessage::InjectGlobals { payload },
200            }),
201            BinaryFrame::Execute {
202                session_id,
203                mode,
204                file_path,
205                bridge_code,
206                post_restore_script,
207                userland_code,
208                high_resolution_time,
209                user_code,
210            } => {
211                if mode > 1 {
212                    return Err(io::Error::new(
213                        io::ErrorKind::InvalidInput,
214                        format!("unknown Execute mode: {mode}"),
215                    ));
216                }
217                Ok(RuntimeCommand::SendToSession {
218                    session_id,
219                    message: SessionMessage::Execute {
220                        mode,
221                        file_path,
222                        bridge_code,
223                        post_restore_script,
224                        userland_code,
225                        high_resolution_time,
226                        user_code,
227                        wasm_module_bytes: None,
228                    },
229                })
230            }
231            BinaryFrame::BridgeResponse {
232                session_id,
233                call_id,
234                status,
235                payload,
236            } => {
237                validate_bridge_response_status(status)?;
238                Ok(RuntimeCommand::SendToSession {
239                    session_id,
240                    message: SessionMessage::BridgeResponse(BridgeResponse {
241                        call_id,
242                        status,
243                        payload,
244                        reservation: None,
245                    }),
246                })
247            }
248            BinaryFrame::StreamEvent {
249                session_id,
250                event_type,
251                payload,
252            } => Ok(RuntimeCommand::SendToSession {
253                session_id,
254                message: SessionMessage::StreamEvent(StreamEvent {
255                    event_type,
256                    payload,
257                }),
258            }),
259            BinaryFrame::TerminateExecution { session_id } => Ok(RuntimeCommand::SendToSession {
260                session_id,
261                message: SessionMessage::TerminateExecution,
262            }),
263            BinaryFrame::WarmSnapshot {
264                bridge_code,
265                userland_code,
266            } => Ok(RuntimeCommand::WarmSnapshot {
267                bridge_code,
268                userland_code,
269            }),
270            BinaryFrame::Authenticate { .. } => Err(io::Error::new(
271                io::ErrorKind::InvalidInput,
272                "Authenticate is not supported by the embedded runtime",
273            )),
274            _ => Err(io::Error::new(
275                io::ErrorKind::InvalidInput,
276                "host-output frames cannot be sent into the embedded runtime",
277            )),
278        }
279    }
280}
281
282impl From<RuntimeEvent> for BinaryFrame {
283    fn from(event: RuntimeEvent) -> Self {
284        match event {
285            RuntimeEvent::BridgeCall {
286                session_id,
287                call_id,
288                method,
289                payload,
290            } => BinaryFrame::BridgeCall {
291                session_id,
292                call_id,
293                method,
294                payload,
295            },
296            RuntimeEvent::ExecutionResult {
297                session_id,
298                exit_code,
299                exports,
300                error,
301            } => BinaryFrame::ExecutionResult {
302                session_id,
303                exit_code,
304                exports,
305                error,
306            },
307            RuntimeEvent::Log {
308                session_id,
309                channel,
310                message,
311            } => BinaryFrame::Log {
312                session_id,
313                channel,
314                message,
315            },
316            RuntimeEvent::StreamCallback {
317                session_id,
318                callback_type,
319                payload,
320            } => BinaryFrame::StreamCallback {
321                session_id,
322                callback_type,
323                payload,
324            },
325        }
326    }
327}
328
329fn non_zero_option(value: u32) -> Option<u32> {
330    if value == 0 {
331        None
332    } else {
333        Some(value)
334    }
335}
336
337pub fn validate_bridge_response_status(status: u8) -> io::Result<()> {
338    if status <= 2 {
339        return Ok(());
340    }
341    Err(io::Error::new(
342        io::ErrorKind::InvalidInput,
343        format!("unknown BridgeResponse status: {status}"),
344    ))
345}
346
347#[cfg(test)]
348mod tests {
349    use super::*;
350
351    #[test]
352    fn rejects_unknown_execute_mode() {
353        let err = RuntimeCommand::try_from(BinaryFrame::Execute {
354            session_id: "s".into(),
355            mode: 2,
356            file_path: "/app/main.mjs".into(),
357            bridge_code: String::new(),
358            post_restore_script: String::new(),
359            userland_code: String::new(),
360            high_resolution_time: false,
361            user_code: String::new(),
362        })
363        .expect_err("unknown execute mode should be rejected");
364
365        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
366        assert!(err.to_string().contains("unknown Execute mode"));
367    }
368
369    #[test]
370    fn rejects_unknown_bridge_response_status() {
371        let err = RuntimeCommand::try_from(BinaryFrame::BridgeResponse {
372            session_id: "s".into(),
373            call_id: 1,
374            status: 3,
375            payload: Vec::new(),
376        })
377        .expect_err("unknown bridge response status should be rejected");
378
379        assert_eq!(err.kind(), io::ErrorKind::InvalidInput);
380        assert!(err.to_string().contains("unknown BridgeResponse status"));
381    }
382
383    #[test]
384    fn accepts_known_bridge_response_statuses() {
385        for status in 0..=2 {
386            let command = RuntimeCommand::try_from(BinaryFrame::BridgeResponse {
387                session_id: "s".into(),
388                call_id: 1,
389                status,
390                payload: Vec::new(),
391            })
392            .expect("known bridge response status should be accepted");
393
394            assert!(matches!(
395                command,
396                RuntimeCommand::SendToSession {
397                    message: SessionMessage::BridgeResponse(BridgeResponse { status: s, .. }),
398                    ..
399                } if s == status
400            ));
401        }
402    }
403}
agentos_v8_runtime/runtime_protocol.rs

agentos_v8_runtime/
runtime_protocol.rs
