agentos_execution/
v8_host.rs

1//! V8 runtime host — manages a shared embedded V8 runtime with session multiplexing.
2
3use crate::v8_ipc::{self, BinaryFrame};
4use agentos_runtime::RuntimeContext;
5use agentos_v8_runtime::embedded_runtime::{
6    shared_embedded_runtime, EmbeddedV8Runtime, EmbeddedV8SessionHandle,
7};
8use agentos_v8_runtime::runtime_protocol::{RuntimeCommand, RuntimeEvent, WarmSessionHint};
9use agentos_v8_runtime::session::RuntimeEventOutputReceiver;
10use std::io::{self, Cursor};
11use std::sync::{Arc, Mutex, OnceLock};
12
13/// V8 polyfill bridge code generated by `build.rs`.
14const V8_BRIDGE_CODE: &str = concat!(
15    include_str!(concat!(env!("OUT_DIR"), "/v8-bridge.js")),
16    "\n",
17    include_str!(concat!(env!("OUT_DIR"), "/v8-bridge-zlib.js"))
18);
19
20/// Manages an embedded V8 runtime with session multiplexing.
21pub struct V8RuntimeHost {
22    shared: Arc<SharedEmbeddedRuntimeClient>,
23}
24
25struct SharedEmbeddedRuntimeClient {
26    runtime: Arc<EmbeddedV8Runtime>,
27}
28
29/// Direct view of the V8 runtime's one bounded per-session output lane. Mapping
30/// the typed in-process event to the legacy binary frame is allocation-local and
31/// no longer requires a relay queue or an OS thread per session.
32pub struct V8SessionFrameReceiver {
33    inner: RuntimeEventOutputReceiver,
34}
35
36impl std::fmt::Debug for V8SessionFrameReceiver {
37    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        formatter
39            .debug_struct("V8SessionFrameReceiver")
40            .finish_non_exhaustive()
41    }
42}
43
44impl V8SessionFrameReceiver {
45    pub fn recv(&self) -> Result<BinaryFrame, flume::RecvError> {
46        self.inner.recv().map(from_runtime_event)
47    }
48
49    pub async fn recv_async(&self) -> Result<BinaryFrame, flume::RecvError> {
50        self.inner.recv_async().await.map(from_runtime_event)
51    }
52}
53
54impl V8RuntimeHost {
55    /// Connect to the process-global embedded V8 runtime client.
56    pub fn spawn(runtime: &RuntimeContext) -> io::Result<Self> {
57        Ok(V8RuntimeHost {
58            shared: shared_embedded_runtime_client(runtime)?,
59        })
60    }
61
62    /// Register a session and return a receiver for its frames.
63    pub fn register_session(
64        &self,
65        session_id: &str,
66        runtime: &RuntimeContext,
67    ) -> io::Result<V8SessionFrameReceiver> {
68        self.shared
69            .runtime
70            .register_session_with_runtime(session_id, runtime)
71            .map(|(inner, _registration)| inner)
72            .map(|inner| V8SessionFrameReceiver { inner })
73    }
74
75    /// Unregister a session.
76    pub fn unregister_session(&self, session_id: &str) {
77        self.shared.runtime.unregister_session(session_id);
78    }
79
80    pub fn create_session(
81        &self,
82        session_id: String,
83        heap_limit_mb: u32,
84        cpu_time_limit_ms: u32,
85        wall_clock_limit_ms: u32,
86        warm_hint: Option<WarmSessionHint>,
87    ) -> io::Result<()> {
88        self.shared.runtime.dispatch(RuntimeCommand::CreateSession {
89            session_id,
90            heap_limit_mb: non_zero_option(heap_limit_mb),
91            cpu_time_limit_ms: non_zero_option(cpu_time_limit_ms),
92            wall_clock_limit_ms: non_zero_option(wall_clock_limit_ms),
93            warm_hint,
94        })
95    }
96
97    pub fn create_session_from_command(&self, command: RuntimeCommand) -> io::Result<()> {
98        self.shared.runtime.dispatch(command)
99    }
100
101    pub fn create_session_from_command_with_runtime(
102        &self,
103        command: RuntimeCommand,
104        runtime: &RuntimeContext,
105        ready_batch_handle_limit: usize,
106        bridge_call_timeout: std::time::Duration,
107    ) -> io::Result<()> {
108        self.shared.runtime.dispatch_create_session_with_runtime(
109            command,
110            runtime.clone(),
111            ready_batch_handle_limit,
112            bridge_call_timeout,
113        )
114    }
115
116    /// Send a frame to the V8 runtime.
117    pub fn send_frame(&self, frame: &BinaryFrame) -> io::Result<()> {
118        self.shared.runtime.dispatch(to_runtime_command(frame)?)
119    }
120
121    /// Get the pre-bundled bridge code (polyfills).
122    pub fn bridge_code() -> &'static str {
123        V8_BRIDGE_CODE
124    }
125
126    /// Pre-build the per-sidecar snapshot for an agent-SDK `userland_code` bundle
127    /// into the process-wide cache, so the FIRST session that uses it is already
128    /// warm (no cold-build penalty on the session-create path). Blocks until the
129    /// snapshot is built; idempotent (a cache hit returns immediately). A no-op for
130    /// empty `userland_code`.
131    pub fn pre_warm_snapshot(&self, userland_code: &str) -> io::Result<()> {
132        if userland_code.is_empty() {
133            return Ok(());
134        }
135        self.shared.runtime.dispatch(RuntimeCommand::WarmSnapshot {
136            bridge_code: Self::bridge_code().to_owned(),
137            userland_code: userland_code.to_owned(),
138        })
139    }
140
141    pub fn pre_warm_workers(&self, userland_code: &str, heap_limit_mb: u32, count: usize) {
142        self.shared.runtime.pre_warm_workers(
143            Self::bridge_code().to_owned(),
144            userland_code.to_owned(),
145            non_zero_option(heap_limit_mb),
146            count,
147        );
148    }
149
150    pub fn seed_default_warm_workers_async(&self) {
151        static DEFAULT_WARM_STARTED: OnceLock<()> = OnceLock::new();
152        let _ = DEFAULT_WARM_STARTED.get_or_init(|| {
153            self.shared.runtime.pre_warm_workers(
154                V8_BRIDGE_CODE.to_owned(),
155                String::new(),
156                None,
157                warm_worker_count(),
158            );
159        });
160    }
161
162    /// True when the process-wide snapshot cache already has this userland
163    /// bundle. This is a lookup only; it never creates a snapshot.
164    pub fn snapshot_ready(&self, userland_code: &str) -> bool {
165        self.shared
166            .runtime
167            .snapshot_ready(Self::bridge_code(), userland_code)
168    }
169
170    /// Kick a process-wide async warm for the wasm runner snapshot. At most one
171    /// warm thread is spawned per process; blocking callers should use
172    /// [`pre_warm_snapshot`](Self::pre_warm_snapshot).
173    pub fn warm_snapshot_async(runtime: &RuntimeContext, userland_code: String) {
174        if userland_code.is_empty() {
175            return;
176        }
177        static WASM_RUNNER_WARM_STARTED: OnceLock<()> = OnceLock::new();
178        let _ = WASM_RUNNER_WARM_STARTED.get_or_init(|| {
179            let requested_bytes = userland_code.len();
180            let runtime_for_job = runtime.clone();
181            if let Err(error) = runtime.blocking().submit(requested_bytes, move || {
182                let result = run_v8_maintenance("agentos-wasm-snapshot-prewarm", move || {
183                    let host = V8RuntimeHost::spawn(&runtime_for_job)?;
184                    host.pre_warm_snapshot(&userland_code)
185                });
186                if let Err(error) = result {
187                    eprintln!("ERR_AGENTOS_V8_MAINTENANCE: wasm snapshot warm failed: {error}");
188                }
189            }) {
190                eprintln!("ERR_AGENTOS_V8_MAINTENANCE: bounded executor rejected warm: {error}");
191            }
192        });
193    }
194
195    /// Create a session handle for sending session-scoped frames and cleanup.
196    pub fn session_handle(&self, session_id: String) -> V8SessionHandle {
197        V8SessionHandle::new(session_id, Arc::clone(&self.shared.runtime))
198    }
199
200    pub fn child_pid(&self) -> u32 {
201        0
202    }
203
204    pub fn is_alive(&mut self) -> io::Result<bool> {
205        Ok(self.shared.runtime.is_alive())
206    }
207
208    #[cfg(test)]
209    fn runtime_ptr(&self) -> usize {
210        Arc::as_ptr(&self.shared.runtime) as usize
211    }
212}
213
214fn non_zero_option(value: u32) -> Option<u32> {
215    (value > 0).then_some(value)
216}
217
218fn warm_worker_count() -> usize {
219    std::env::var("AGENTOS_V8_WARM_ISOLATES")
220        .ok()
221        .and_then(|value| value.parse::<usize>().ok())
222        .unwrap_or(2)
223}
224
225/// A handle to a single V8 session within the shared runtime.
226/// Provides methods for sending frames specific to this session.
227pub struct V8SessionHandle {
228    inner: EmbeddedV8SessionHandle,
229}
230
231impl std::fmt::Debug for V8SessionHandle {
232    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
233        f.debug_struct("V8SessionHandle")
234            .field("session_id", &self.inner.session_id())
235            .finish()
236    }
237}
238
239impl V8SessionHandle {
240    pub fn new(session_id: String, runtime: Arc<EmbeddedV8Runtime>) -> Self {
241        Self {
242            inner: runtime.session_handle(session_id),
243        }
244    }
245
246    /// Send a bridge response back to the V8 isolate.
247    pub fn send_bridge_response(
248        &self,
249        call_id: u64,
250        status: u8,
251        payload: Vec<u8>,
252    ) -> io::Result<()> {
253        self.inner.send_bridge_response(call_id, status, payload)
254    }
255
256    /// Send a stream event to the V8 isolate (stdin data, timer, etc.).
257    pub fn send_stream_event(&self, event_type: &str, payload: Vec<u8>) -> io::Result<()> {
258        self.inner.send_stream_event(event_type, payload)
259    }
260
261    /// Publish coalesced readiness for a registered sidecar capability. This
262    /// carries no subsystem payload; the guest drain targets durable sidecar
263    /// state by capability identity.
264    pub fn publish_readiness(
265        &self,
266        capability_id: u64,
267        capability_generation: u64,
268        flags: agentos_runtime::readiness::ReadyFlags,
269    ) -> io::Result<()> {
270        self.inner
271            .publish_readiness(capability_id, capability_generation, flags)
272    }
273
274    pub fn remove_readiness(
275        &self,
276        capability_id: u64,
277        capability_generation: u64,
278    ) -> io::Result<()> {
279        self.inner
280            .remove_readiness(capability_id, capability_generation)
281    }
282
283    pub fn set_application_read_interest(
284        &self,
285        capability_id: u64,
286        capability_generation: u64,
287        enabled: bool,
288    ) -> io::Result<()> {
289        self.inner
290            .set_application_read_interest(capability_id, capability_generation, enabled)
291    }
292
293    pub fn publish_signal(&self, signal: i32) -> io::Result<()> {
294        self.inner.publish_signal(signal)
295    }
296
297    pub fn publish_timer(&self, timer_id: u64) -> io::Result<()> {
298        self.inner.publish_timer(timer_id)
299    }
300
301    /// Install a direct module-source reader on this session's V8 thread so module
302    /// loads read source directly instead of round-tripping the bridge.
303    pub fn set_module_reader(
304        &self,
305        reader: Box<dyn agentos_v8_runtime::execution::GuestModuleReader>,
306    ) -> io::Result<()> {
307        self.inner.set_module_reader(reader)
308    }
309
310    /// Execute bridge code + user code in this V8 session without routing through
311    /// the legacy binary frame encode/decode path.
312    #[allow(clippy::too_many_arguments)] // mirrors the CreateSession frame
313    pub fn execute(
314        &self,
315        mode: u8,
316        file_path: String,
317        bridge_code: String,
318        post_restore_script: String,
319        userland_code: String,
320        high_resolution_time: bool,
321        user_code: String,
322        wasm_module_bytes: Option<Arc<Vec<u8>>>,
323    ) -> io::Result<()> {
324        self.inner.execute(
325            mode,
326            file_path,
327            bridge_code,
328            post_restore_script,
329            userland_code,
330            high_resolution_time,
331            user_code,
332            wasm_module_bytes,
333        )
334    }
335
336    /// Terminate execution in this session.
337    pub fn terminate(&self) -> io::Result<()> {
338        self.inner.terminate()
339    }
340
341    /// Suspend guest execution while preserving the V8 stack and session state.
342    pub fn pause(&self) -> io::Result<()> {
343        self.inner.pause()
344    }
345
346    /// Resume a session previously suspended with [`Self::pause`].
347    pub fn resume(&self) -> io::Result<()> {
348        self.inner.resume()
349    }
350
351    /// Destroy this session in the embedded runtime and remove its receiver.
352    pub fn destroy(&self) -> io::Result<()> {
353        self.inner.destroy()
354    }
355
356    pub fn session_id(&self) -> &str {
357        self.inner.session_id()
358    }
359}
360
361impl Clone for V8SessionHandle {
362    fn clone(&self) -> Self {
363        Self {
364            inner: self.inner.clone(),
365        }
366    }
367}
368
369/// Pre-build the per-sidecar snapshot for an agent-SDK `userland_code` bundle into
370/// the process-wide cache, so the FIRST session that uses it is already warm. Uses
371/// the shared embedded runtime directly (no per-call host lifecycle). Blocks until
372/// built; idempotent (cache hit returns immediately); no-op for empty input.
373/// Eagerly initialize the process-wide embedded V8 runtime (and its process-lifetime
374/// platform owner) before accepting VM work. Calling this at sidecar startup keeps
375/// initialization failures on the entrypoint, although correctness no longer
376/// depends on the first caller itself being a long-lived thread.
377pub fn ensure_runtime_initialized(runtime: &RuntimeContext) -> io::Result<()> {
378    shared_embedded_runtime_client(runtime).map(|_| ())
379}
380
381pub fn pre_warm_agent_snapshot(runtime: &RuntimeContext, userland_code: &str) -> io::Result<()> {
382    if userland_code.is_empty() {
383        return Ok(());
384    }
385    let userland = userland_code.to_owned();
386    let runtime = runtime.clone();
387    run_v8_maintenance("agentos-snapshot-prewarm", move || {
388        let client = shared_embedded_runtime_client(&runtime)?;
389        client.runtime.dispatch(RuntimeCommand::WarmSnapshot {
390            bridge_code: V8_BRIDGE_CODE.to_owned(),
391            userland_code: userland,
392        })
393    })
394}
395
396/// V8 snapshot construction is the one admitted ephemeral maintenance-thread
397/// exception. A process-wide mutex serializes it, and every thread is joined
398/// before the next job can start so thread-local V8 state cannot leak.
399fn run_v8_maintenance<T: Send + 'static>(
400    thread_name: &str,
401    operation: impl FnOnce() -> io::Result<T> + Send + 'static,
402) -> io::Result<T> {
403    static V8_MAINTENANCE_LOCK: Mutex<()> = Mutex::new(());
404    let _exclusive = V8_MAINTENANCE_LOCK
405        .lock()
406        .map_err(|_| io::Error::other("V8 maintenance lock poisoned"))?;
407    // AGENTOS_THREAD_SITE: serialized-v8-maintenance
408    let handle = std::thread::Builder::new()
409        .name(thread_name.to_owned())
410        .spawn(operation)?;
411    handle
412        .join()
413        .map_err(|_| io::Error::other("V8 maintenance thread panicked"))?
414}
415
416fn shared_embedded_runtime_client(
417    runtime_context: &RuntimeContext,
418) -> io::Result<Arc<SharedEmbeddedRuntimeClient>> {
419    static SHARED_RUNTIME: OnceLock<Arc<SharedEmbeddedRuntimeClient>> = OnceLock::new();
420    static SHARED_RUNTIME_INIT_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
421
422    if let Some(shared) = SHARED_RUNTIME.get() {
423        return Ok(Arc::clone(shared));
424    }
425
426    let _guard = SHARED_RUNTIME_INIT_LOCK
427        .lock()
428        .expect("shared embedded runtime init lock poisoned");
429    if let Some(shared) = SHARED_RUNTIME.get() {
430        return Ok(Arc::clone(shared));
431    }
432
433    let shared = Arc::new(SharedEmbeddedRuntimeClient {
434        runtime: shared_embedded_runtime(runtime_context.clone())?,
435    });
436    let _ = SHARED_RUNTIME.set(Arc::clone(&shared));
437    Ok(shared)
438}
439
440fn to_runtime_command(frame: &BinaryFrame) -> io::Result<RuntimeCommand> {
441    let bytes = v8_ipc::encode_frame(frame)?;
442    let runtime_frame = agentos_v8_runtime::ipc_binary::read_frame(&mut Cursor::new(bytes))?;
443    RuntimeCommand::try_from(runtime_frame)
444}
445
446fn from_runtime_event(event: RuntimeEvent) -> BinaryFrame {
447    match event {
448        RuntimeEvent::BridgeCall {
449            session_id,
450            call_id,
451            method,
452            payload,
453        } => BinaryFrame::BridgeCall {
454            session_id,
455            call_id,
456            method,
457            payload,
458        },
459        RuntimeEvent::ExecutionResult {
460            session_id,
461            exit_code,
462            exports,
463            error,
464        } => BinaryFrame::ExecutionResult {
465            session_id,
466            exit_code,
467            exports,
468            error: error.map(from_runtime_execution_error),
469        },
470        RuntimeEvent::Log {
471            session_id,
472            channel,
473            message,
474        } => BinaryFrame::Log {
475            session_id,
476            channel,
477            message,
478        },
479        RuntimeEvent::StreamCallback {
480            session_id,
481            callback_type,
482            payload,
483        } => BinaryFrame::StreamCallback {
484            session_id,
485            callback_type,
486            payload,
487        },
488    }
489}
490
491fn from_runtime_execution_error(
492    error: agentos_v8_runtime::ipc_binary::ExecutionErrorBin,
493) -> v8_ipc::ExecutionErrorBin {
494    v8_ipc::ExecutionErrorBin {
495        error_type: error.error_type,
496        message: error.message,
497        stack: error.stack,
498        code: error.code,
499    }
500}
501
502#[cfg(test)]
503mod tests {
504    use super::*;
505    use std::sync::atomic::{AtomicU64, Ordering};
506
507    static NEXT_TEST_SESSION_ID: AtomicU64 = AtomicU64::new(1);
508
509    fn next_session_id() -> String {
510        format!(
511            "embedded-runtime-host-{}",
512            NEXT_TEST_SESSION_ID.fetch_add(1, Ordering::Relaxed)
513        )
514    }
515
516    fn test_runtime_context() -> RuntimeContext {
517        agentos_runtime::SidecarRuntime::process(&agentos_runtime::RuntimeConfig::default())
518            .expect("test process runtime")
519            .context()
520    }
521
522    #[test]
523    fn embedded_runtime_host_reuses_shared_runtime_service() {
524        let runtime = test_runtime_context();
525        let first = V8RuntimeHost::spawn(&runtime).expect("spawn V8 runtime host");
526        let second = V8RuntimeHost::spawn(&runtime).expect("spawn V8 runtime host");
527        assert_eq!(
528            first.runtime_ptr(),
529            second.runtime_ptr(),
530            "V8 runtime hosts should reuse the same embedded runtime service"
531        );
532    }
533
534    #[test]
535    fn embedded_runtime_host_create_destroy_recycles_session_ids() {
536        let runtime = test_runtime_context();
537        let host = V8RuntimeHost::spawn(&runtime).expect("spawn V8 runtime host");
538        let session_id = next_session_id();
539
540        let _first_receiver = host
541            .register_session(&session_id, &runtime)
542            .expect("register session output");
543        host.send_frame(&BinaryFrame::CreateSession {
544            session_id: session_id.clone(),
545            heap_limit_mb: 0,
546            cpu_time_limit_ms: 0,
547            wall_clock_limit_ms: 0,
548        })
549        .expect("create embedded runtime session");
550
551        let duplicate_error = host
552            .send_frame(&BinaryFrame::CreateSession {
553                session_id: session_id.clone(),
554                heap_limit_mb: 0,
555                cpu_time_limit_ms: 0,
556                wall_clock_limit_ms: 0,
557            })
558            .expect_err("duplicate session ids should be rejected");
559        assert_eq!(duplicate_error.kind(), io::ErrorKind::Other);
560
561        host.session_handle(session_id.clone())
562            .destroy()
563            .expect("destroy embedded runtime session");
564
565        let _second_receiver = host
566            .register_session(&session_id, &runtime)
567            .expect("re-register session output");
568        host.send_frame(&BinaryFrame::CreateSession {
569            session_id: session_id.clone(),
570            heap_limit_mb: 0,
571            cpu_time_limit_ms: 0,
572            wall_clock_limit_ms: 0,
573        })
574        .expect("recreate embedded runtime session");
575
576        host.session_handle(session_id)
577            .destroy()
578            .expect("destroy recreated session");
579    }
580}
agentos_execution/v8_host.rs

agentos_execution/
v8_host.rs
