agentos_v8_runtime/

session.rs

// Session management: create/destroy sessions with V8 isolates on dedicated threads

#[cfg(not(test))]
use std::collections::BTreeMap;
use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::atomic::AtomicU64;
use std::sync::{Arc, Condvar, Mutex, OnceLock};
use std::thread;
#[cfg(not(test))]
use std::time::{Duration, Instant};

#[cfg(not(test))]
use agentos_bridge::queue_tracker::{warn_limit_exhausted, TrackedLimit};
use agentos_bridge::{bridge_contract, BridgeCallConvention};
use crossbeam_channel::{Receiver, Sender};

use crate::execution;
#[cfg(not(test))]
use crate::host_call::{BridgeCallContext, ChannelRuntimeEventSender};
use crate::host_call::{CallIdRouter, SharedCallIdCounter};
use crate::ipc::ExecutionError;
#[cfg(not(test))]
use crate::ipc_binary::ExecutionErrorBin;
use crate::runtime_protocol::{
    BridgeResponse, RuntimeEvent, SessionMessage, StreamEvent, WarmSessionHint,
};
use crate::snapshot::{snapshot_cache_key, SnapshotCache, SnapshotCacheKey};
#[cfg(not(test))]
use crate::{bridge, isolate, snapshot};

/// Commands sent to a session thread
pub enum SessionCommand {
    /// Shut down the session and destroy the isolate
    Shutdown,
    /// Forward a typed session message to the session thread for processing
    Message(SessionMessage),
    /// Install a direct module-source reader on the session thread. Carried as a
    /// live object over the in-process command channel (NOT a serialized frame),
    /// so subsequent module loads on this thread read source directly instead of
    /// round-tripping the bridge. Sent just before an Execute message.
    SetModuleReader(Box<dyn crate::execution::GuestModuleReader>),
}

#[cfg(not(test))]
type SharedIsolateHandle = Arc<Mutex<Option<v8::IsolateHandle>>>;
#[cfg(test)]
type SharedIsolateHandle = Arc<Mutex<Option<()>>>;

/// Sender for typed runtime events produced by session threads.
pub type RuntimeEventSender = crossbeam_channel::Sender<RuntimeEventEnvelope>;

#[derive(Debug, Clone, PartialEq)]
pub struct RuntimeEventEnvelope {
    pub output_generation: Option<u64>,
    pub event: RuntimeEvent,
}

const LATE_TERMINATE_EXECUTION_ERROR_CODE: &str = "ERR_LATE_TERMINATE_EXECUTION";
const LATE_STREAM_EVENT_ERROR_CODE: &str = "ERR_LATE_STREAM_EVENT";
const LATE_BRIDGE_RESPONSE_ERROR_CODE: &str = "ERR_LATE_BRIDGE_RESPONSE";
const DEFERRED_COMMAND_LIMIT_ERROR_CODE: &str = "ERR_SESSION_DEFERRED_COMMAND_LIMIT";
const SESSION_COMMAND_CHANNEL_CAPACITY: usize = 256;
const MAX_DEFERRED_SESSION_COMMANDS: usize = SESSION_COMMAND_CHANNEL_CAPACITY;
const MAX_DEFERRED_SYNC_MESSAGES: usize = SESSION_COMMAND_CHANNEL_CAPACITY;

#[derive(Clone, Debug, Eq, Hash, PartialEq)]
struct WarmPoolKey {
    snapshot_key_digest: SnapshotCacheKey,
    heap_limit_mb: u32,
}

struct ParkedWorker {
    assignment_tx: Sender<SessionAssignment>,
    join_handle: thread::JoinHandle<()>,
}

#[derive(Default)]
struct WarmWorkerPoolState {
    workers: HashMap<WarmPoolKey, Vec<ParkedWorker>>,
    refilling: HashSet<WarmPoolKey>,
}

#[derive(Default)]
struct WarmWorkerPool {
    state: Mutex<WarmWorkerPoolState>,
}

struct SessionAssignment {
    heap_limit_mb: Option<u32>,
    cpu_time_limit_ms: Option<u32>,
    wall_clock_limit_ms: Option<u32>,
    rx: Receiver<SessionCommand>,
    slot_control: SlotControl,
    max_concurrency: usize,
    event_tx: RuntimeEventSender,
    call_id_router: CallIdRouter,
    shared_call_id: SharedCallIdCounter,
    snapshot_cache: Arc<SnapshotCache>,
    isolate_handle: SharedIsolateHandle,
    execution_abort: SharedExecutionAbort,
    session_id: String,
    output_generation: Option<u64>,
}

#[cfg(not(test))]
struct PrecreatedIsolate {
    isolate: v8::OwnedIsolate,
    context: v8::Global<v8::Context>,
    bridge_code: String,
    userland_code: String,
}

#[cfg(test)]
struct PrecreatedIsolate;

#[cfg(not(test))]
#[derive(Default)]
struct V8SessionPhaseStats {
    calls: u64,
    total_ns: u128,
    max_ns: u128,
}

#[cfg(not(test))]
static V8_SESSION_PHASES: OnceLock<Mutex<BTreeMap<String, V8SessionPhaseStats>>> = OnceLock::new();

#[cfg(not(test))]
fn v8_session_phases_enabled() -> bool {
    std::env::var("AGENTOS_V8_SESSION_PHASES").as_deref() == Ok("1")
}

#[cfg(not(test))]
fn record_v8_session_phase(stage: &str, elapsed: Duration) {
    if !v8_session_phases_enabled() {
        return;
    }
    let phases = V8_SESSION_PHASES.get_or_init(|| Mutex::new(BTreeMap::new()));
    let Ok(mut phases) = phases.lock() else {
        return;
    };
    let stats = phases.entry(stage.to_string()).or_default();
    stats.calls += 1;
    let elapsed_ns = elapsed.as_nanos();
    stats.total_ns += elapsed_ns;
    stats.max_ns = stats.max_ns.max(elapsed_ns);

    let Some(path) = std::env::var_os("AGENTOS_V8_SESSION_PHASES_FILE") else {
        return;
    };
    let mut output = String::new();
    for (stage, stats) in phases.iter() {
        let total_us = stats.total_ns / 1_000;
        let avg_us = if stats.calls == 0 {
            0
        } else {
            total_us / u128::from(stats.calls)
        };
        let max_us = stats.max_ns / 1_000;
        output.push_str(&format!(
            "stage={stage} calls={} total_us={total_us} avg_us={avg_us} max_us={max_us}\n",
            stats.calls
        ));
    }
    let _ = std::fs::write(path, output);
}

#[cfg(not(test))]
fn record_warm_worker_hit() {
    record_v8_session_phase("warm_worker_hit", Duration::ZERO);
}

#[cfg(test)]
fn record_warm_worker_hit() {}

#[cfg(not(test))]
fn record_warm_worker_miss() {
    record_v8_session_phase("warm_worker_miss", Duration::ZERO);
}

#[cfg(test)]
fn record_warm_worker_miss() {}

fn warm_worker_capacity_per_key() -> usize {
    std::env::var("AGENTOS_V8_WARM_ISOLATES")
        .ok()
        .and_then(|value| value.parse::<usize>().ok())
        .unwrap_or(2)
}

fn effective_heap_limit_mb(heap_limit_mb: Option<u32>) -> u32 {
    heap_limit_mb.unwrap_or(crate::isolate::DEFAULT_HEAP_LIMIT_MB)
}

fn warm_pool_key(
    bridge_code: &str,
    userland_code: &str,
    heap_limit_mb: Option<u32>,
) -> WarmPoolKey {
    WarmPoolKey {
        snapshot_key_digest: snapshot_cache_key(
            bridge_code,
            (!userland_code.is_empty()).then_some(userland_code),
        ),
        heap_limit_mb: effective_heap_limit_mb(heap_limit_mb),
    }
}

fn warm_key_prefix(key: &WarmPoolKey) -> String {
    key.snapshot_key_digest[..4]
        .iter()
        .map(|byte| format!("{byte:02x}"))
        .collect()
}

impl WarmWorkerPool {
    fn claim(&self, key: &WarmPoolKey) -> Option<ParkedWorker> {
        let mut state = self.state.lock().expect("warm worker pool lock poisoned");
        state.workers.get_mut(key).and_then(Vec::pop)
    }

    fn shutdown_handles(&self) -> Vec<thread::JoinHandle<()>> {
        let mut state = self.state.lock().expect("warm worker pool lock poisoned");
        state.refilling.clear();
        state
            .workers
            .drain()
            .flat_map(|(_, workers)| workers)
            .map(|worker| {
                drop(worker.assignment_tx);
                worker.join_handle
            })
            .collect()
    }

    fn ensure_count(
        self: &Arc<Self>,
        snapshot_cache: Arc<SnapshotCache>,
        slot_control: SlotControl,
        bridge_code: String,
        userland_code: String,
        heap_limit_mb: Option<u32>,
        requested_count: usize,
    ) {
        let capacity = warm_worker_capacity_per_key();
        if capacity == 0 || requested_count == 0 {
            return;
        }

        let target_count = requested_count.min(capacity);
        let key = warm_pool_key(&bridge_code, &userland_code, heap_limit_mb);
        {
            let mut state = self.state.lock().expect("warm worker pool lock poisoned");
            let current = state.workers.get(&key).map_or(0, Vec::len);
            if current >= target_count || state.refilling.contains(&key) {
                return;
            }
            state.refilling.insert(key.clone());
        }

        let pool = Arc::clone(self);
        let spawn_key = key.clone();
        let _ = thread::Builder::new()
            .name(String::from("secure-exec-v8-warm-refill"))
            .spawn(move || {
                pool.refill_until(
                    snapshot_cache,
                    slot_control,
                    spawn_key,
                    bridge_code,
                    userland_code,
                    heap_limit_mb,
                    target_count,
                );
            })
            .map_err(|error| {
                eprintln!("agentos-v8-runtime: warm worker refill spawn failed: {error}");
                self.state
                    .lock()
                    .expect("warm worker pool lock poisoned")
                    .refilling
                    .remove(&key);
            });
    }

    // Internal pool-refill plumbing; args mirror the parked-worker construction.
    #[allow(clippy::too_many_arguments)]
    fn refill_until(
        &self,
        snapshot_cache: Arc<SnapshotCache>,
        slot_control: SlotControl,
        key: WarmPoolKey,
        bridge_code: String,
        userland_code: String,
        heap_limit_mb: Option<u32>,
        target_count: usize,
    ) {
        loop {
            let capacity = warm_worker_capacity_per_key();
            if capacity == 0 {
                break;
            }
            let desired = target_count.min(capacity);
            {
                let state = self.state.lock().expect("warm worker pool lock poisoned");
                if state.workers.get(&key).map_or(0, Vec::len) >= desired {
                    break;
                }
            }

            let Some(worker) = spawn_warm_worker(
                Arc::clone(&snapshot_cache),
                Arc::clone(&slot_control),
                key.clone(),
                bridge_code.clone(),
                userland_code.clone(),
                heap_limit_mb,
            ) else {
                break;
            };

            let mut state = self.state.lock().expect("warm worker pool lock poisoned");
            let workers = state.workers.entry(key.clone()).or_default();
            if workers.len() >= desired {
                drop(worker.assignment_tx);
                let _ = worker.join_handle.join();
                break;
            }
            workers.push(worker);
            eprintln!(
                "agentos-v8-runtime: warm worker refilled key={} heap={} pool_size={}",
                warm_key_prefix(&key),
                key.heap_limit_mb,
                workers.len()
            );
        }

        self.state
            .lock()
            .expect("warm worker pool lock poisoned")
            .refilling
            .remove(&key);
    }
}

#[cfg(not(test))]
fn spawn_warm_worker(
    snapshot_cache: Arc<SnapshotCache>,
    slot_control: SlotControl,
    key: WarmPoolKey,
    bridge_code: String,
    userland_code: String,
    heap_limit_mb: Option<u32>,
) -> Option<ParkedWorker> {
    let (assignment_tx, assignment_rx) = crossbeam_channel::bounded::<SessionAssignment>(1);
    let (ready_tx, ready_rx) = crossbeam_channel::bounded::<Result<(), String>>(1);
    let worker_bridge_code = bridge_code.clone();
    let worker_userland_code = userland_code.clone();
    let join_handle = match thread::Builder::new()
        .name(String::from("secure-exec-v8-warm-worker"))
        .spawn(move || {
            let precreated = precreate_warm_isolate(
                snapshot_cache,
                slot_control,
                worker_bridge_code,
                worker_userland_code,
                heap_limit_mb,
            );
            match precreated {
                Ok(precreated) => {
                    let _ = ready_tx.send(Ok(()));
                    if let Ok(assignment) = assignment_rx.recv() {
                        session_thread(assignment, Some(precreated));
                    }
                }
                Err(error) => {
                    let _ = ready_tx.send(Err(error));
                }
            }
        }) {
        Ok(handle) => handle,
        Err(error) => {
            eprintln!("agentos-v8-runtime: warm worker spawn failed: {error}");
            return None;
        }
    };

    match ready_rx.recv() {
        Ok(Ok(())) => Some(ParkedWorker {
            assignment_tx,
            join_handle,
        }),
        Ok(Err(error)) => {
            eprintln!(
                "agentos-v8-runtime: warm worker refill failed key={} heap={}: {error}",
                warm_key_prefix(&key),
                key.heap_limit_mb
            );
            let _ = join_handle.join();
            None
        }
        Err(error) => {
            eprintln!(
                "agentos-v8-runtime: warm worker refill failed key={} heap={}: {error}",
                warm_key_prefix(&key),
                key.heap_limit_mb
            );
            let _ = join_handle.join();
            None
        }
    }
}

#[cfg(test)]
fn spawn_warm_worker(
    _snapshot_cache: Arc<SnapshotCache>,
    _slot_control: SlotControl,
    _key: WarmPoolKey,
    _bridge_code: String,
    _userland_code: String,
    _heap_limit_mb: Option<u32>,
) -> Option<ParkedWorker> {
    None
}

#[cfg(not(test))]
fn precreate_warm_isolate(
    snapshot_cache: Arc<SnapshotCache>,
    slot_control: SlotControl,
    bridge_code: String,
    userland_code: String,
    heap_limit_mb: Option<u32>,
) -> Result<PrecreatedIsolate, String> {
    isolate::init_v8_platform();
    let snapshot_blob = snapshot_cache.get_or_create_with_userland(
        &bridge_code,
        (!userland_code.is_empty()).then_some(userland_code.as_str()),
    )?;
    let snapshot_blob = (*snapshot_blob).clone();
    let _idle_slots = lock_idle_slots(&slot_control);
    let mut isolate = snapshot::create_isolate_from_snapshot(snapshot_blob, heap_limit_mb);
    isolate.set_host_import_module_dynamically_callback(execution::dynamic_import_callback);
    isolate.set_host_initialize_import_meta_object_callback(execution::import_meta_object_callback);
    let context = isolate::create_context(&mut isolate);
    Ok(PrecreatedIsolate {
        isolate,
        context,
        bridge_code,
        userland_code,
    })
}

#[cfg(not(test))]
fn lock_idle_slots(slot_control: &SlotControl) -> std::sync::MutexGuard<'_, usize> {
    let (lock, cvar) = &**slot_control;
    let mut count = lock.lock().unwrap();
    while *count != 0 {
        count = cvar.wait(count).unwrap();
    }
    count
}

/// Normalize an opt-in CPU-time budget: `Some(0)` means "disabled" and folds to
/// `None` so the CPU-budget watchdog is NOT armed. The runtime layer does not
/// invent a default here: secure-exec sidecar VM executions pass the typed
/// `limits.jsRuntime.cpuTimeLimitMs` default, while lower-level callers can pass
/// `None`/`0` deliberately.
fn normalize_cpu_time_limit_ms(cpu_time_limit_ms: Option<u32>) -> Option<u32> {
    cpu_time_limit_ms.filter(|budget_ms| *budget_ms > 0)
}

/// Normalize an opt-in WALL-CLOCK backstop: `Some(0)` means "disabled" and folds
/// to `None` so the wall-clock `TimeoutGuard` is NOT armed. There is no default —
/// when the caller passes `None`/`0`, the guest runs with no wall-clock limit
/// (opt-in by design, so long-lived ACP adapters are never killed by a default).
/// This is INDEPENDENT of the CPU-time budget: setting one does not arm the other.
fn normalize_wall_clock_limit_ms(wall_clock_limit_ms: Option<u32>) -> Option<u32> {
    wall_clock_limit_ms.filter(|limit_ms| *limit_ms > 0)
}

/// Internal entry for a running session
struct SessionEntry {
    /// Output receiver generation current when this session was created.
    output_generation: Option<u64>,
    /// Channel to send commands to the session thread
    tx: Sender<SessionCommand>,
    /// Thread join handle
    join_handle: Option<thread::JoinHandle<()>>,
    /// Thread-safe V8 isolate handle for out-of-band termination.
    #[cfg_attr(test, allow(dead_code))]
    isolate_handle: SharedIsolateHandle,
    /// Current execution abort handle used to wake sync bridge waits.
    execution_abort: SharedExecutionAbort,
}

/// Deferred shutdown work for a session that has already been removed from
/// the manager. `finish()` joins the session thread and clears any call
/// routes the thread registered while shutting down. Callers must release
/// the SessionManager lock before calling `finish()`. Joining under the lock
/// deadlocks: the dispatch thread needs the lock to drain the event channel,
/// and the joined thread can be parked on a full event channel send.
pub struct SessionShutdown {
    session_id: String,
    join_handle: Option<thread::JoinHandle<()>>,
    call_id_router: CallIdRouter,
}

impl SessionShutdown {
    pub fn finish(mut self) {
        if let Some(handle) = self.join_handle.take() {
            let _ = handle.join();
        }
        self.call_id_router
            .lock()
            .expect("call_id router lock poisoned")
            .retain(|_, routed_session_id| routed_session_id != &self.session_id);
    }
}

/// Concurrency slot tracker shared across session threads
type SlotControl = Arc<(Mutex<usize>, Condvar)>;

/// Shared deferred message queue for non-BridgeResponse frames consumed by
/// sync bridge calls. The event loop drains these before blocking on the channel.
pub(crate) type DeferredQueue = Arc<Mutex<VecDeque<SessionMessage>>>;

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ExecutionAbortReason {
    /// Caller explicitly terminated the execution (e.g. session destroy).
    Terminated,
    /// The opt-in WALL-CLOCK backstop (`TimeoutGuard`) elapsed. Counts elapsed
    /// real time INCLUDING idle/await, so it can cap a guest that blocks/awaits
    /// indefinitely. Armed only when `limits.jsRuntime.wallClockLimitMs` is set;
    /// independent of the CPU-time budget.
    #[cfg_attr(test, allow(dead_code))]
    WallClockTimedOut,
    /// The TRUE CPU-TIME budget (`CpuBudgetGuard`) was exhausted by active JS CPU.
    #[cfg_attr(test, allow(dead_code))]
    CpuBudgetExceeded,
}

struct ExecutionAbortState {
    sender: Option<crossbeam_channel::Sender<()>>,
    reason: Option<ExecutionAbortReason>,
}

pub(crate) struct SharedExecutionAbort(Arc<Mutex<Option<ExecutionAbortState>>>);

impl Clone for SharedExecutionAbort {
    fn clone(&self) -> Self {
        Self(Arc::clone(&self.0))
    }
}

/// Create a new empty deferred queue.
pub(crate) fn new_deferred_queue() -> DeferredQueue {
    Arc::new(Mutex::new(VecDeque::new()))
}

pub(crate) fn new_execution_abort() -> SharedExecutionAbort {
    SharedExecutionAbort(Arc::new(Mutex::new(None)))
}

pub(crate) struct ActiveExecutionAbort {
    shared: SharedExecutionAbort,
}

impl ActiveExecutionAbort {
    pub(crate) fn arm(shared: &SharedExecutionAbort) -> (Self, crossbeam_channel::Receiver<()>) {
        let (tx, rx) = crossbeam_channel::bounded::<()>(0);
        let mut guard = shared.0.lock().unwrap();
        *guard = Some(ExecutionAbortState {
            sender: Some(tx),
            reason: None,
        });
        (
            Self {
                shared: shared.clone(),
            },
            rx,
        )
    }
}

impl Drop for ActiveExecutionAbort {
    fn drop(&mut self) {
        *self.shared.0.lock().unwrap() = None;
    }
}

pub(crate) fn signal_execution_abort(shared: &SharedExecutionAbort, reason: ExecutionAbortReason) {
    if let Some(state) = shared.0.lock().unwrap().as_mut() {
        state.reason.get_or_insert(reason);
        state.sender.take();
    }
}

#[cfg(not(test))]
fn execution_abort_reason(shared: &SharedExecutionAbort) -> Option<ExecutionAbortReason> {
    shared
        .0
        .lock()
        .unwrap()
        .as_ref()
        .and_then(|state| state.reason)
}

/// Manages V8 sessions with concurrency limiting.
/// Each session runs on a dedicated OS thread with its own V8 isolate.
pub struct SessionManager {
    sessions: HashMap<String, SessionEntry>,
    max_concurrency: usize,
    slot_control: SlotControl,
    /// Typed runtime event sender shared across session threads.
    event_tx: RuntimeEventSender,
    /// Call_id → session_id routing table for BridgeResponse dispatch
    call_id_router: CallIdRouter,
    /// Shared call_id counter — all sessions use this to generate globally unique
    /// call_ids, preventing collisions in the call_id_router
    shared_call_id: SharedCallIdCounter,
    /// Shared snapshot cache for fast isolate creation from pre-compiled bridge code
    snapshot_cache: Arc<SnapshotCache>,
    /// Ready-to-claim isolate workers keyed by snapshot digest and heap cap.
    warm_pool: Arc<WarmWorkerPool>,
}

impl SessionManager {
    pub fn new(
        max_concurrency: usize,
        event_tx: RuntimeEventSender,
        call_id_router: CallIdRouter,
        snapshot_cache: Arc<SnapshotCache>,
    ) -> Self {
        SessionManager {
            sessions: HashMap::new(),
            max_concurrency,
            slot_control: Arc::new((Mutex::new(0), Condvar::new())),
            event_tx,
            call_id_router,
            shared_call_id: Arc::new(AtomicU64::new(1)),
            snapshot_cache,
            warm_pool: Arc::new(WarmWorkerPool::default()),
        }
    }

    /// Get the snapshot cache for pre-warming from WarmSnapshot messages.
    #[allow(dead_code)]
    pub fn snapshot_cache(&self) -> &Arc<SnapshotCache> {
        &self.snapshot_cache
    }

    pub fn pre_warm_workers(
        &self,
        bridge_code: String,
        userland_code: String,
        heap_limit_mb: Option<u32>,
        count: usize,
    ) {
        self.warm_pool.ensure_count(
            Arc::clone(&self.snapshot_cache),
            Arc::clone(&self.slot_control),
            bridge_code,
            userland_code,
            heap_limit_mb,
            count,
        );
    }

    /// Create a new session.
    /// Spawns a dedicated thread with a V8 isolate. If max concurrency is
    /// reached, the session thread will block until a slot becomes available.
    pub fn create_session(
        &mut self,
        session_id: String,
        heap_limit_mb: Option<u32>,
        cpu_time_limit_ms: Option<u32>,
        wall_clock_limit_ms: Option<u32>,
    ) -> Result<(), String> {
        self.create_session_with_output_generation(
            session_id,
            heap_limit_mb,
            cpu_time_limit_ms,
            wall_clock_limit_ms,
            None,
            None,
        )
    }

    pub fn create_session_with_output_generation(
        &mut self,
        session_id: String,
        heap_limit_mb: Option<u32>,
        cpu_time_limit_ms: Option<u32>,
        wall_clock_limit_ms: Option<u32>,
        output_generation: Option<u64>,
        warm_hint: Option<WarmSessionHint>,
    ) -> Result<(), String> {
        if self.sessions.contains_key(&session_id) {
            return Err(format!("session {} already exists", session_id));
        }

        let cpu_time_limit_ms = normalize_cpu_time_limit_ms(cpu_time_limit_ms);
        let wall_clock_limit_ms = normalize_wall_clock_limit_ms(wall_clock_limit_ms);
        let (tx, rx) = crossbeam_channel::bounded(SESSION_COMMAND_CHANNEL_CAPACITY);
        let isolate_handle = Arc::new(Mutex::new(None));
        let execution_abort = new_execution_abort();
        let assignment = SessionAssignment {
            heap_limit_mb,
            cpu_time_limit_ms,
            wall_clock_limit_ms,
            rx,
            slot_control: Arc::clone(&self.slot_control),
            max_concurrency: self.max_concurrency,
            event_tx: self.event_tx.clone(),
            call_id_router: Arc::clone(&self.call_id_router),
            shared_call_id: Arc::clone(&self.shared_call_id),
            snapshot_cache: Arc::clone(&self.snapshot_cache),
            isolate_handle: Arc::clone(&isolate_handle),
            execution_abort: execution_abort.clone(),
            session_id: session_id.clone(),
            output_generation,
        };

        let join_handle = match self.claim_warm_worker(warm_hint.as_ref(), assignment) {
            Ok((join_handle, true)) => {
                if let Some(hint) = warm_hint {
                    self.warm_pool.ensure_count(
                        Arc::clone(&self.snapshot_cache),
                        Arc::clone(&self.slot_control),
                        hint.bridge_code,
                        hint.userland_code,
                        hint.heap_limit_mb,
                        warm_worker_capacity_per_key(),
                    );
                }
                join_handle
            }
            Ok((join_handle, false)) => join_handle,
            Err(assignment) => spawn_session_thread(assignment)
                .map_err(|e| format!("failed to spawn session thread: {}", e))?,
        };

        self.sessions.insert(
            session_id,
            SessionEntry {
                output_generation,
                tx,
                join_handle: Some(join_handle),
                isolate_handle,
                execution_abort,
            },
        );

        Ok(())
    }

    // The Err variant intentionally carries the whole SessionAssignment back to
    // the caller for the fallback spawn path — it is moved, not copied.
    #[allow(clippy::result_large_err)]
    fn claim_warm_worker(
        &self,
        warm_hint: Option<&WarmSessionHint>,
        assignment: SessionAssignment,
    ) -> Result<(thread::JoinHandle<()>, bool), SessionAssignment> {
        let Some(hint) = warm_hint else {
            return Err(assignment);
        };
        if warm_worker_capacity_per_key() == 0 {
            record_warm_worker_miss();
            return Err(assignment);
        }

        let key = warm_pool_key(&hint.bridge_code, &hint.userland_code, hint.heap_limit_mb);
        let Some(worker) = self.warm_pool.claim(&key) else {
            record_warm_worker_miss();
            eprintln!(
                "agentos-v8-runtime: warm worker pool-empty key={} heap={}",
                warm_key_prefix(&key),
                key.heap_limit_mb
            );
            return Err(assignment);
        };

        match worker.assignment_tx.send(assignment) {
            Ok(()) => {
                record_warm_worker_hit();
                eprintln!(
                    "agentos-v8-runtime: warm worker claimed key={} heap={}",
                    warm_key_prefix(&key),
                    key.heap_limit_mb
                );
                Ok((worker.join_handle, true))
            }
            Err(error) => {
                record_warm_worker_miss();
                let _ = worker.join_handle.join();
                Err(error.0)
            }
        }
    }

    pub fn destroy_session_if_output_generation(
        &mut self,
        session_id: &str,
        output_generation: u64,
    ) -> Result<bool, String> {
        match self.begin_destroy_session_if_output_generation(session_id, output_generation)? {
            Some(shutdown) => {
                shutdown.finish();
                Ok(true)
            }
            None => Ok(false),
        }
    }

    pub fn begin_destroy_session_if_output_generation(
        &mut self,
        session_id: &str,
        output_generation: u64,
    ) -> Result<Option<SessionShutdown>, String> {
        if self
            .sessions
            .get(session_id)
            .is_none_or(|entry| entry.output_generation != Some(output_generation))
        {
            return Ok(None);
        }

        self.begin_destroy_session(session_id).map(Some)
    }

    pub fn detach_session_if_output_generation(
        &mut self,
        session_id: &str,
        output_generation: u64,
    ) -> Result<bool, String> {
        if self
            .sessions
            .get(session_id)
            .is_none_or(|entry| entry.output_generation != Some(output_generation))
        {
            return Ok(false);
        }

        self.detach_session(session_id)?;
        Ok(true)
    }

    fn detach_session(&mut self, session_id: &str) -> Result<(), String> {
        let entry = self
            .sessions
            .get(session_id)
            .ok_or_else(|| format!("session {} does not exist", session_id))?;

        #[cfg(not(test))]
        if let Some(handle) = entry
            .isolate_handle
            .lock()
            .ok()
            .and_then(|guard| guard.as_ref().cloned())
        {
            handle.terminate_execution();
        }
        signal_execution_abort(&entry.execution_abort, ExecutionAbortReason::Terminated);
        self.clear_call_routes_for_session(session_id);
        let mut entry = self.sessions.remove(session_id).unwrap();
        let _ = entry.tx.try_send(SessionCommand::Shutdown);
        drop(entry.tx);
        let _ = entry.join_handle.take();
        Ok(())
    }

    /// Destroy a session inline. Joins the session thread before returning, so
    /// this must not be called while a shared lock on the manager is held. Lock
    /// holders use `begin_destroy_session` and call `finish()` after unlocking.
    pub fn destroy_session(&mut self, session_id: &str) -> Result<(), String> {
        self.begin_destroy_session(session_id)?.finish();
        Ok(())
    }

    /// First phase of destroying a session: terminate execution, signal abort,
    /// send shutdown, clear call routes, and remove the entry. The returned
    /// shutdown joins the session thread and must be finished after the
    /// SessionManager lock is released.
    pub fn begin_destroy_session(&mut self, session_id: &str) -> Result<SessionShutdown, String> {
        if !self.sessions.contains_key(session_id) {
            return Err(format!("session {} does not exist", session_id));
        }

        self.clear_call_routes_for_session(session_id);
        let mut entry = self
            .sessions
            .remove(session_id)
            .expect("checked session exists");

        #[cfg(not(test))]
        if let Some(handle) = entry
            .isolate_handle
            .lock()
            .ok()
            .and_then(|guard| guard.as_ref().cloned())
        {
            handle.terminate_execution();
        }
        signal_execution_abort(&entry.execution_abort, ExecutionAbortReason::Terminated);
        // Send shutdown, then drop the entry (and with it the sender) so the
        // session thread's rx.recv() returns Err if Shutdown was consumed by
        // an inner loop.
        let _ = entry.tx.try_send(SessionCommand::Shutdown);
        let join_handle = entry.join_handle.take();
        drop(entry);
        Ok(SessionShutdown {
            session_id: session_id.to_owned(),
            join_handle,
            call_id_router: Arc::clone(&self.call_id_router),
        })
    }

    pub(crate) fn take_session_shutdown_handles(&mut self) -> Vec<thread::JoinHandle<()>> {
        self.call_id_router
            .lock()
            .expect("call_id router lock poisoned")
            .clear();

        let mut handles: Vec<_> = self
            .sessions
            .drain()
            .filter_map(|(_, mut entry)| {
                #[cfg(not(test))]
                if let Some(handle) = entry
                    .isolate_handle
                    .lock()
                    .ok()
                    .and_then(|guard| guard.as_ref().cloned())
                {
                    handle.terminate_execution();
                }
                signal_execution_abort(&entry.execution_abort, ExecutionAbortReason::Terminated);
                let _ = entry.tx.try_send(SessionCommand::Shutdown);
                drop(entry.tx);
                entry.join_handle.take()
            })
            .collect();
        handles.extend(self.warm_pool.shutdown_handles());
        handles
    }

    pub(crate) fn clear_call_route(&self, call_id: u64) {
        self.call_id_router
            .lock()
            .expect("call_id router lock poisoned")
            .remove(&call_id);
    }

    fn clear_call_routes_for_session(&self, session_id: &str) {
        self.call_id_router
            .lock()
            .expect("call_id router lock poisoned")
            .retain(|_, routed_session_id| routed_session_id != session_id);
    }

    /// Resolve a session's command sender and apply message side effects that
    /// must happen under the manager lock (isolate termination, abort signal).
    /// The caller sends on the returned channel after releasing the lock so a
    /// full command channel cannot block the manager mutex.
    pub fn session_command_sender(
        &self,
        session_id: &str,
        msg: &SessionMessage,
    ) -> Result<Sender<SessionCommand>, String> {
        let entry = self
            .sessions
            .get(session_id)
            .ok_or_else(|| format!("session {} does not exist", session_id))?;

        #[cfg(not(test))]
        if matches!(msg, SessionMessage::TerminateExecution) {
            if let Some(handle) = entry
                .isolate_handle
                .lock()
                .ok()
                .and_then(|guard| guard.as_ref().cloned())
            {
                handle.terminate_execution();
            }
        }
        if matches!(msg, SessionMessage::TerminateExecution) {
            signal_execution_abort(&entry.execution_abort, ExecutionAbortReason::Terminated);
        }

        Ok(entry.tx.clone())
    }

    /// Get a session's command sender without a message (used for control commands
    /// like SetModuleReader that aren't a SessionMessage). Dispatch-thread only.
    pub fn session_sender(&self, session_id: &str) -> Result<Sender<SessionCommand>, String> {
        self.sessions
            .get(session_id)
            .map(|entry| entry.tx.clone())
            .ok_or_else(|| format!("session {} does not exist", session_id))
    }

    /// Send a message to a session. Blocks on the session command channel, so
    /// this must not be called while a shared lock on the manager is held.
    pub fn send_to_session(&self, session_id: &str, msg: SessionMessage) -> Result<(), String> {
        let sender = self.session_command_sender(session_id, &msg)?;
        sender
            .send(SessionCommand::Message(msg))
            .map_err(|e| format!("session thread disconnected: {}", e))
    }

    /// Destroy a set of sessions inline, ignoring sessions that were already
    /// removed. Joins session threads, so this must not be called while a
    /// shared lock on the manager is held.
    pub fn destroy_sessions<I>(&mut self, session_ids: I)
    where
        I: IntoIterator<Item = String>,
    {
        for shutdown in self.begin_destroy_sessions(session_ids) {
            shutdown.finish();
        }
    }

    /// Begin destroying a set of sessions, ignoring sessions that were already
    /// removed. Finish each returned shutdown after releasing the manager lock.
    pub fn begin_destroy_sessions<I>(&mut self, session_ids: I) -> Vec<SessionShutdown>
    where
        I: IntoIterator<Item = String>,
    {
        session_ids
            .into_iter()
            .filter_map(|sid| self.begin_destroy_session(&sid).ok())
            .collect()
    }

    /// Number of registered sessions (including those waiting for a slot).
    #[allow(dead_code)]
    pub fn session_count(&self) -> usize {
        self.sessions.len()
    }

    /// Return all session IDs.
    #[allow(dead_code)]
    pub fn all_sessions(&self) -> Vec<String> {
        self.sessions.keys().cloned().collect()
    }

    /// Number of sessions that have acquired a concurrency slot.
    #[allow(dead_code)]
    pub fn active_slot_count(&self) -> usize {
        let (lock, _) = &*self.slot_control;
        *lock.lock().unwrap()
    }

    /// Get the call_id routing table for BridgeResponse dispatch.
    pub fn call_id_router(&self) -> &CallIdRouter {
        &self.call_id_router
    }
}

/// Send a typed runtime event without re-serializing it on the session thread.
#[cfg(not(test))]
fn send_event_with_generation(
    event_tx: &RuntimeEventSender,
    output_generation: Option<u64>,
    event: RuntimeEvent,
) {
    if let Err(error) = event_tx.send(RuntimeEventEnvelope {
        output_generation,
        event,
    }) {
        eprintln!("failed to send runtime event: {error}");
    }
}

fn send_late_message_warning(
    event_tx: &RuntimeEventSender,
    session_id: &str,
    output_generation: Option<u64>,
    error_code: &str,
    detail: String,
) {
    let warning = RuntimeEvent::Log {
        session_id: session_id.to_string(),
        channel: 1,
        message: format!("[{error_code}] {detail}"),
    };
    if let Err(error) = event_tx.send(RuntimeEventEnvelope {
        output_generation,
        event: warning,
    }) {
        eprintln!("failed to send late-session warning: {error}");
    }
}

fn handle_late_session_message(
    event_tx: &RuntimeEventSender,
    session_id: &str,
    output_generation: Option<u64>,
    message: SessionMessage,
) {
    match message {
        SessionMessage::BridgeResponse(BridgeResponse {
            call_id,
            status,
            payload,
        }) => send_late_message_warning(
            event_tx,
            session_id,
            output_generation,
            LATE_BRIDGE_RESPONSE_ERROR_CODE,
            format!(
                "dropping BridgeResponse after execution completed (call_id={call_id}, status={status}, payload_len={})",
                payload.len()
            ),
        ),
        SessionMessage::StreamEvent(StreamEvent {
            event_type,
            payload,
        }) => {
            // Timer and socket-readiness events are wake hints, not data; both
            // race execution completion by design (edge-triggered readiness can
            // fire as the guest finishes) and carry nothing to lose, so drop
            // them silently instead of warning.
            if event_type == "timer" || event_type == "net_socket" {
                return;
            }
            send_late_message_warning(
                event_tx,
                session_id,
                output_generation,
                LATE_STREAM_EVENT_ERROR_CODE,
                format!(
                    "dropping StreamEvent after execution completed (event_type={event_type}, payload_len={})",
                    payload.len()
                ),
            )
        }
        SessionMessage::TerminateExecution => send_late_message_warning(
            event_tx,
            session_id,
            output_generation,
            LATE_TERMINATE_EXECUTION_ERROR_CODE,
            String::from("dropping TerminateExecution after execution completed"),
        ),
        SessionMessage::InjectGlobals { .. } | SessionMessage::Execute { .. } => {}
    }
}

fn defer_session_command_before_slot(
    deferred_commands: &mut VecDeque<SessionCommand>,
    event_tx: &RuntimeEventSender,
    session_id: &str,
    output_generation: Option<u64>,
    command: SessionCommand,
) -> bool {
    if deferred_commands.len() < MAX_DEFERRED_SESSION_COMMANDS {
        deferred_commands.push_back(command);
        return true;
    }

    send_late_message_warning(
        event_tx,
        session_id,
        output_generation,
        DEFERRED_COMMAND_LIMIT_ERROR_CODE,
        format!(
            "dropping queued session before slot acquisition because deferred command queue exceeded limit of {MAX_DEFERRED_SESSION_COMMANDS}"
        ),
    );
    false
}

#[cfg(not(test))]
fn install_wasm_module_bytes_global<'s>(scope: &mut v8::HandleScope<'s>, bytes: &[u8]) -> bool {
    let global = scope.get_current_context().global(scope);
    let Some(name) = v8::String::new(scope, "__agentOSWasmModuleBytes") else {
        return false;
    };
    let len = bytes.len();
    let backing_store = v8::ArrayBuffer::new_backing_store_from_bytes(bytes.to_vec());
    let array_buffer = v8::ArrayBuffer::with_backing_store(scope, &backing_store.make_shared());
    let Some(bytes_value) = v8::Uint8Array::new(scope, array_buffer, 0, len) else {
        return false;
    };
    global.set(scope, name.into(), bytes_value.into()).is_some()
}

/// Session thread: acquires a concurrency slot, defers V8 isolate creation
/// to first Execute (when bridge code is known for snapshot lookup), and
/// processes commands until shutdown.
fn spawn_session_thread(assignment: SessionAssignment) -> std::io::Result<thread::JoinHandle<()>> {
    let name_prefix = if assignment.session_id.len() > 8 {
        assignment.session_id[..8].to_string()
    } else {
        assignment.session_id.clone()
    };
    thread::Builder::new()
        .name(format!("session-{}", name_prefix))
        .spawn(move || session_thread(assignment, None))
}

#[allow(clippy::too_many_arguments)]
fn session_thread(
    assignment: SessionAssignment,
    #[cfg_attr(test, allow(unused_variables))] precreated_isolate: Option<PrecreatedIsolate>,
) {
    let SessionAssignment {
        heap_limit_mb,
        cpu_time_limit_ms,
        wall_clock_limit_ms,
        rx,
        slot_control,
        max_concurrency,
        event_tx,
        call_id_router,
        shared_call_id,
        snapshot_cache,
        isolate_handle,
        execution_abort,
        session_id,
        output_generation,
    } = assignment;
    #[cfg(test)]
    let _ = (
        heap_limit_mb,
        cpu_time_limit_ms,
        wall_clock_limit_ms,
        call_id_router,
        shared_call_id,
        snapshot_cache,
        isolate_handle,
        execution_abort,
    );

    // Acquire concurrency slot, but keep polling the session channel so a queued
    // session can still shut down cleanly before it ever gets a slot.
    let mut deferred_commands = VecDeque::new();
    let acquired_slot = {
        let (lock, cvar) = &*slot_control;
        let mut count = lock.lock().unwrap();
        loop {
            if *count < max_concurrency {
                *count += 1;
                break true;
            }

            let (next_count, _) = cvar
                .wait_timeout(count, std::time::Duration::from_millis(50))
                .unwrap();
            count = next_count;

            match rx.try_recv() {
                Ok(SessionCommand::Shutdown)
                | Err(crossbeam_channel::TryRecvError::Disconnected) => {
                    break false;
                }
                Ok(command) => {
                    if !defer_session_command_before_slot(
                        &mut deferred_commands,
                        &event_tx,
                        &session_id,
                        output_generation,
                        command,
                    ) {
                        break false;
                    }
                }
                Err(crossbeam_channel::TryRecvError::Empty) => {}
            }
        }
    };

    if !acquired_slot {
        return;
    }

    // Capture THIS session thread's per-thread CPU clock once. The clock id is
    // stable for the thread's lifetime and can be polled from the watchdog
    // thread; this is what lets the CPU-budget guard measure active JS CPU time
    // (excluding idle/await) without running on the execution thread itself.
    // Guest JS always runs on this thread, so this clock is the execution clock.
    #[cfg(all(not(test), unix))]
    let exec_thread_cpu_clock = crate::timeout::current_thread_cpu_clock();
    #[cfg(all(not(test), not(unix)))]
    let exec_thread_cpu_clock: Option<crate::timeout::ThreadCpuClock> = None;

    // Isolate creation is normally deferred to first Execute (when bridge code is
    // known for snapshot cache lookup). A claimed warm worker enters here with
    // the snapshot isolate already created on this same thread.
    #[cfg(not(test))]
    let (
        mut v8_isolate,
        mut _v8_context,
        mut from_snapshot,
        mut isolate_bridge_code,
        mut isolate_userland_code,
    ) = match precreated_isolate {
        Some(precreated) => (
            Some(precreated.isolate),
            Some(precreated.context),
            true,
            Some(precreated.bridge_code),
            Some(precreated.userland_code),
        ),
        None => (None, None, false, None, None),
    };

    #[cfg(not(test))]
    let mut pending = bridge::PendingPromises::new();

    // Store latest InjectGlobals V8 payload for re-injection into fresh contexts
    #[cfg(not(test))]
    let mut last_globals_payload: Option<Vec<u8>> = None;

    // Bridge code cache for V8 code caching across executions
    #[cfg(not(test))]
    let mut bridge_cache: Option<execution::BridgeCodeCache> = None;

    // Cached bridge code string to skip resending over IPC
    #[cfg(not(test))]
    let mut last_bridge_code: Option<String> = None;
    // Cached agent-SDK userland bundle (same 0-length = use cached convention).
    #[cfg(not(test))]
    let mut last_userland_code: Option<String> = None;

    // A session can reuse its isolate across Executes only while the effective
    // bridge code stays the same. Fresh contexts cloned from a snapshot inherit
    // the snapshot's bridge IIFE, so a bridge-code change must rebuild the
    // isolate before the next execution or the session will keep restoring the
    // old snapshot forever. The userland bundle is part of the same guard.
    #[cfg(not(test))]
    let mut high_resolution_time_origin = Instant::now();

    #[cfg(not(test))]
    if let Some(iso) = v8_isolate.as_mut() {
        *isolate_handle
            .lock()
            .expect("session isolate handle lock poisoned") = Some(iso.thread_safe_handle());
    }

    // Process commands until shutdown or channel close
    loop {
        let next_command = if let Some(command) = deferred_commands.pop_front() {
            Ok(command)
        } else {
            rx.recv()
        };

        match next_command {
            Ok(SessionCommand::Shutdown) | Err(_) => break,
            Ok(SessionCommand::SetModuleReader(reader)) => {
                execution::install_session_guest_reader(Some(reader));
            }
            Ok(SessionCommand::Message(msg)) => match msg {
                SessionMessage::InjectGlobals { payload } => {
                    #[cfg(not(test))]
                    {
                        // Store V8-serialized config for injection into fresh context at Execute time
                        last_globals_payload = Some(payload);
                    }
                    #[cfg(test)]
                    {
                        let _ = payload;
                    }
                }
                SessionMessage::Execute {
                    mode,
                    file_path,
                    bridge_code,
                    post_restore_script,
                    userland_code,
                    high_resolution_time,
                    user_code,
                    wasm_module_bytes,
                } => {
                    // `userland_code` is consumed only by the non-test snapshot
                    // path below; keep it bound (without a warning) under `test`.
                    #[cfg(test)]
                    let _ = &userland_code;
                    #[cfg(test)]
                    let _ = high_resolution_time;
                    #[cfg(test)]
                    let _ = &wasm_module_bytes;
                    #[cfg(not(test))]
                    {
                        let session_id = session_id.clone();
                        // Use cached bridge code when host sends empty (0-length = use cached)
                        let should_update_cached_bridge_code = !bridge_code.is_empty();
                        let effective_bridge_code = if bridge_code.is_empty() {
                            last_bridge_code.as_deref().unwrap_or("").to_string()
                        } else {
                            bridge_code
                        };
                        // Same 0-length = use-cached convention for the userland bundle.
                        let should_update_cached_userland_code = !userland_code.is_empty();
                        let effective_userland_code = if userland_code.is_empty() {
                            last_userland_code.as_deref().unwrap_or("").to_string()
                        } else {
                            userland_code
                        };

                        if let Err(message) =
                            snapshot::validate_bridge_code_size(&effective_bridge_code)
                        {
                            let result_frame = RuntimeEvent::ExecutionResult {
                                session_id,
                                exit_code: 1,
                                exports: None,
                                error: Some(ExecutionErrorBin {
                                    error_type: "Error".into(),
                                    message,
                                    stack: String::new(),
                                    code: snapshot::V8_BRIDGE_CODE_LIMIT_ERROR_CODE.into(),
                                }),
                            };
                            send_event_with_generation(&event_tx, output_generation, result_frame);
                            continue;
                        }

                        if should_update_cached_bridge_code {
                            last_bridge_code = Some(effective_bridge_code.clone());
                        }
                        if should_update_cached_userland_code {
                            last_userland_code = Some(effective_userland_code.clone());
                        }

                        if v8_isolate.is_some()
                            && (isolate_bridge_code.as_deref()
                                != Some(effective_bridge_code.as_str())
                                || isolate_userland_code.as_deref()
                                    != Some(effective_userland_code.as_str()))
                        {
                            *isolate_handle
                                .lock()
                                .expect("session isolate handle lock poisoned") = None;
                            // Reset pending promise-resolver Globals BEFORE this
                            // isolate is dropped. The registry is reused across
                            // isolate rebuilds, and a prior execution that was
                            // terminated early (Shutdown / timeout-abort) can
                            // leave resolvers registered, so they would otherwise
                            // outlive the isolate that created them.
                            reset_pending_promises(&mut pending);
                            drop(_v8_context.take());
                            isolate::drop_isolate(v8_isolate.take());
                            from_snapshot = false;
                            isolate_bridge_code = None;
                            isolate_userland_code = None;
                        }

                        // Deferred isolate creation: create on first Execute using snapshot cache
                        if v8_isolate.is_none() {
                            isolate::init_v8_platform();
                            // The snapshot captures the bridge AND (when present) the
                            // agent-SDK userland bundle, keyed process-wide by both, so
                            // the SDK is evaluated once per sidecar and reused here.
                            let phase_start = Instant::now();
                            let snapshot_blob = match snapshot_cache.get_or_create_with_userland(
                                &effective_bridge_code,
                                (!effective_userland_code.is_empty())
                                    .then_some(effective_userland_code.as_str()),
                            ) {
                                Ok(blob) => Some(blob),
                                Err(message) => {
                                    // Snapshot creation runs in a helper subprocess; if
                                    // that fails (unsupported platform, spawn failure),
                                    // degrade to a fresh isolate that evaluates the
                                    // bridge in-context rather than failing the session.
                                    eprintln!(
                                        "agentos-v8-runtime: snapshot creation failed, \
                                         falling back to fresh isolate: {message}"
                                    );
                                    None
                                }
                            };
                            record_v8_session_phase("snapshot_get", phase_start.elapsed());
                            let mut iso = match snapshot_blob {
                                Some(blob) => {
                                    from_snapshot = true;
                                    eprintln!(
                                        "agentos-v8-runtime: restored session isolate from_snapshot=true"
                                    );
                                    let phase_start = Instant::now();
                                    // rusty_v8 0.130's CreateParams::snapshot_blob
                                    // takes owned 'static data, so this copy remains
                                    // per exec until the API can accept cached bytes.
                                    let snapshot_blob = (*blob).clone();
                                    record_v8_session_phase("blob_clone", phase_start.elapsed());
                                    let phase_start = Instant::now();
                                    let isolate = snapshot::create_isolate_from_snapshot(
                                        snapshot_blob,
                                        heap_limit_mb,
                                    );
                                    record_v8_session_phase("isolate_new", phase_start.elapsed());
                                    isolate
                                }
                                None => {
                                    from_snapshot = false;
                                    let phase_start = Instant::now();
                                    let isolate = isolate::create_isolate(heap_limit_mb);
                                    record_v8_session_phase("isolate_new", phase_start.elapsed());
                                    isolate
                                }
                            };
                            iso.set_host_import_module_dynamically_callback(
                                execution::dynamic_import_callback,
                            );
                            iso.set_host_initialize_import_meta_object_callback(
                                execution::import_meta_object_callback,
                            );
                            high_resolution_time_origin = Instant::now();
                            *isolate_handle
                                .lock()
                                .expect("session isolate handle lock poisoned") =
                                Some(iso.thread_safe_handle());
                            let ctx = isolate::create_context(&mut iso);
                            _v8_context = Some(ctx);
                            v8_isolate = Some(iso);
                            isolate_bridge_code = Some(effective_bridge_code.clone());
                            isolate_userland_code = Some(effective_userland_code.clone());
                        }

                        let iso = v8_isolate.as_mut().unwrap();
                        iso.cancel_terminate_execution();

                        // Create execution context: Context::new on a snapshot-restored
                        // isolate gives a fresh clone of the snapshot's default context
                        // (bridge IIFE already executed, all infrastructure set up).
                        // On a non-snapshot isolate, this gives a blank context.
                        let exec_context = isolate::create_context(iso);

                        if high_resolution_time {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            execution::install_high_resolution_time_global(
                                scope,
                                &high_resolution_time_origin as *const Instant,
                            );
                        }

                        // Inject globals from last InjectGlobals payload
                        if let Some(ref payload) = last_globals_payload {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            if let Err(error) =
                                execution::inject_globals_from_payload(scope, payload)
                            {
                                let result_frame = RuntimeEvent::ExecutionResult {
                                    session_id,
                                    exit_code: 1,
                                    exports: None,
                                    error: Some(ExecutionErrorBin {
                                        error_type: error.error_type,
                                        message: error.message,
                                        stack: error.stack,
                                        code: error.code.unwrap_or_default(),
                                    }),
                                };
                                send_event_with_generation(
                                    &event_tx,
                                    output_generation,
                                    result_frame,
                                );
                                continue;
                            }
                        }

                        // Arm a per-execution abort channel so timeouts and external
                        // terminate requests can unblock sync bridge waits.
                        let (_active_execution_abort, abort_rx) =
                            ActiveExecutionAbort::arm(&execution_abort);

                        // Create deferred queue for sync bridge call filtering
                        let deferred_queue = new_deferred_queue();

                        // Create BridgeCallContext with channel sender (no shared mutex)
                        let channel_rx = ChannelResponseReceiver::with_abort(
                            rx.clone(),
                            abort_rx.clone(),
                            Arc::clone(&deferred_queue),
                        );
                        let bridge_ctx = BridgeCallContext::with_receiver(
                            Box::new(ChannelRuntimeEventSender::new(
                                event_tx.clone(),
                                output_generation,
                            )),
                            Box::new(channel_rx),
                            session_id.clone(),
                            Arc::clone(&call_id_router),
                            Arc::clone(&shared_call_id),
                        );

                        // Replace stub bridge functions with real session-local ones
                        // (on snapshot context) or register from scratch (on fresh context).
                        // Both paths use the same function — global.set() works for both.
                        let _sync_store;
                        let _async_store;
                        let sync_bridge_fns = sync_bridge_fns();
                        let async_bridge_fns = async_bridge_fns();
                        {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);

                            (_sync_store, _async_store) = bridge::replace_bridge_fns(
                                scope,
                                &bridge_ctx as *const BridgeCallContext,
                                &pending as *const bridge::PendingPromises,
                                sync_bridge_fns,
                                async_bridge_fns,
                            );
                        }

                        // Run post-restore init script (config, mutable state reset)
                        // after bridge fn replacement but before user code
                        if !post_restore_script.is_empty() {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            let (prs_code, prs_err) =
                                execution::run_init_script(scope, &post_restore_script);
                            if prs_code != 0 {
                                let result_frame = RuntimeEvent::ExecutionResult {
                                    session_id,
                                    exit_code: prs_code,
                                    exports: None,
                                    error: prs_err.map(|e| ExecutionErrorBin {
                                        error_type: e.error_type,
                                        message: e.message,
                                        stack: e.stack,
                                        code: e.code.unwrap_or_default(),
                                    }),
                                };
                                send_event_with_generation(
                                    &event_tx,
                                    output_generation,
                                    result_frame,
                                );
                                continue;
                            }
                        }

                        if let Some(wasm_module_bytes) = wasm_module_bytes.as_ref() {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            if !install_wasm_module_bytes_global(scope, wasm_module_bytes) {
                                let result_frame = RuntimeEvent::ExecutionResult {
                                    session_id,
                                    exit_code: 1,
                                    exports: None,
                                    error: Some(ExecutionErrorBin {
                                        error_type: "Error".into(),
                                        message: "failed to install __agentOSWasmModuleBytes"
                                            .into(),
                                        stack: String::new(),
                                        code: String::new(),
                                    }),
                                };
                                send_event_with_generation(
                                    &event_tx,
                                    output_generation,
                                    result_frame,
                                );
                                continue;
                            }
                        }

                        // Arm the TRUE CPU-TIME budget watchdog before running
                        // guest code when the caller passes a nonzero
                        // `limits.jsRuntime.cpuTimeLimitMs` (normalized: `0`/unset =>
                        // `None` => not armed at this runtime layer). The sidecar
                        // supplies the bounded default for VM executions.
                        //
                        // The watchdog counts ACTIVE JS CPU only (idle/await
                        // excluded) by polling the execution thread's CPU clock, so
                        // a guest that mostly awaits is NOT killed by it. The
                        // INDEPENDENT wall-clock backstop (armed just below) covers
                        // the idle/await case when the operator opts into it.
                        let mut cpu_budget_guard = match cpu_time_limit_ms {
                            Some(budget_ms) => {
                                // Enforcing a CPU budget requires the execution
                                // thread's CPU clock captured at session start. If
                                // it is unavailable we cannot honor the operator's
                                // requested cap — surface that rather than silently
                                // running uncapped.
                                let cpu_clock = match exec_thread_cpu_clock {
                                    Some(clock) => clock,
                                    None => {
                                        let result_frame = RuntimeEvent::ExecutionResult {
                                            session_id,
                                            exit_code: 1,
                                            exports: None,
                                            error: Some(ExecutionErrorBin {
                                                error_type: "Error".into(),
                                                message: format!(
                                                    "{}: per-thread CPU clock unavailable; cannot enforce limits.jsRuntime.cpuTimeLimitMs",
                                                    crate::timeout::CPU_BUDGET_GUARD_START_ERROR_CODE
                                                ),
                                                stack: String::new(),
                                                code: crate::timeout::CPU_BUDGET_GUARD_START_ERROR_CODE
                                                    .into(),
                                            }),
                                        };
                                        send_event_with_generation(
                                            &event_tx,
                                            output_generation,
                                            result_frame,
                                        );
                                        continue;
                                    }
                                };
                                let handle = iso.thread_safe_handle();
                                match crate::timeout::CpuBudgetGuard::new(
                                    budget_ms,
                                    cpu_clock,
                                    handle,
                                    execution_abort.clone(),
                                ) {
                                    Ok(guard) => Some(guard),
                                    Err(message) => {
                                        let result_frame = RuntimeEvent::ExecutionResult {
                                            session_id,
                                            exit_code: 1,
                                            exports: None,
                                            error: Some(ExecutionErrorBin {
                                                error_type: "Error".into(),
                                                message,
                                                stack: String::new(),
                                                code:
                                                    crate::timeout::CPU_BUDGET_GUARD_START_ERROR_CODE
                                                        .into(),
                                            }),
                                        };
                                        send_event_with_generation(
                                            &event_tx,
                                            output_generation,
                                            result_frame,
                                        );
                                        continue;
                                    }
                                }
                            }
                            _ => None,
                        };

                        // Arm the INDEPENDENT, opt-in WALL-CLOCK backstop alongside
                        // the CPU budget. Unlike the CPU budget, this counts elapsed
                        // real time INCLUDING idle/await, so it can cap a guest that
                        // blocks or awaits indefinitely. Armed only when the operator
                        // opts in via `limits.jsRuntime.wallClockLimitMs` (normalized:
                        // `0`/unset => `None` => not armed => NO wall-clock limit, so
                        // long-lived ACP adapters are never killed by a default).
                        // Whichever guard fires first calls `terminate_execution` and
                        // records its abort reason; the result frame reports which.
                        let mut wall_clock_guard = match wall_clock_limit_ms {
                            Some(limit_ms) => {
                                let handle = iso.thread_safe_handle();
                                match crate::timeout::TimeoutGuard::with_execution_abort(
                                    limit_ms,
                                    handle,
                                    execution_abort.clone(),
                                ) {
                                    Ok(guard) => Some(guard),
                                    Err(message) => {
                                        let result_frame = RuntimeEvent::ExecutionResult {
                                            session_id,
                                            exit_code: 1,
                                            exports: None,
                                            error: Some(ExecutionErrorBin {
                                                error_type: "Error".into(),
                                                message,
                                                stack: String::new(),
                                                code:
                                                    crate::timeout::TIMEOUT_GUARD_START_ERROR_CODE
                                                        .into(),
                                            }),
                                        };
                                        send_event_with_generation(
                                            &event_tx,
                                            output_generation,
                                            result_frame,
                                        );
                                        continue;
                                    }
                                }
                            }
                            _ => None,
                        };

                        // On snapshot-restored context, skip bridge IIFE (already in
                        // snapshot) and run user code only. On fresh context, run full
                        // bridge code + user code as before.
                        let bridge_code_for_exec = if from_snapshot {
                            ""
                        } else {
                            &effective_bridge_code
                        };
                        let file_path_opt = if file_path.is_empty() {
                            None
                        } else {
                            Some(file_path.as_str())
                        };
                        let phase_start = Instant::now();
                        let (mut code, mut exports, mut error) = if mode == 0 {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            let (c, e) = execution::execute_script_with_options(
                                scope,
                                Some(&bridge_ctx),
                                bridge_code_for_exec,
                                &user_code,
                                file_path_opt,
                                &mut bridge_cache,
                            );
                            (c, None, e)
                        } else {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            execution::execute_module(
                                scope,
                                &bridge_ctx,
                                bridge_code_for_exec,
                                &user_code,
                                file_path_opt,
                                &mut bridge_cache,
                            )
                        };

                        // Re-check async ESM completion once immediately so
                        // pure-microtask top-level await settles without
                        // needing a bridge event-loop round-trip.
                        if mode != 0 && error.is_none() {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            if let Some((next_code, next_exports, next_error)) =
                                execution::finalize_pending_module_evaluation(scope)
                            {
                                code = next_code;
                                exports = next_exports;
                                error = next_error;
                            }
                        }
                        record_v8_session_phase("user_code_execute", phase_start.elapsed());

                        // Run event loop while bridge work or async ESM
                        // evaluation is still pending. For ESM modules (mode != 0),
                        // always enter the event loop even if no pending promises
                        // are visible yet — the module body may have registered
                        // timers, stdin listeners, or child_process handles that
                        // need event loop pumping to deliver their callbacks.
                        let should_enter_event_loop = !pending.is_empty()
                            || execution::has_pending_module_evaluation()
                            || execution::has_pending_script_evaluation()
                            || !deferred_queue.lock().unwrap().is_empty();
                        let event_loop_status = if should_enter_event_loop {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            run_event_loop(
                                scope,
                                &rx,
                                &pending,
                                Some(&abort_rx),
                                Some(&deferred_queue),
                            )
                        } else {
                            EventLoopStatus::Completed
                        };

                        let mut terminated =
                            matches!(event_loop_status, EventLoopStatus::Terminated);
                        if let EventLoopStatus::Failed(next_code, next_error) = event_loop_status {
                            code = next_code;
                            error = Some(next_error);
                        }

                        // Finalize any entry-module top-level await that was
                        // waiting on bridge-driven async work (timers/network).
                        if !terminated && mode != 0 && error.is_none() {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            if let Some((next_code, next_exports, next_error)) =
                                execution::finalize_pending_module_evaluation(scope)
                            {
                                code = next_code;
                                exports = next_exports;
                                error = next_error;
                            }
                        }

                        // Keep the session alive while handles (timers, child
                        // processes, stdin listeners) are active. Long-lived
                        // ACP adapters often run as plain scripts, so this
                        // cannot be limited to ESM entrypoints.
                        if !terminated && error.is_none() {
                            // Phase 1: call _waitForActiveHandles() to register a pending promise
                            {
                                let scope = &mut v8::HandleScope::new(iso);
                                let ctx = v8::Local::new(scope, &exec_context);
                                let scope = &mut v8::ContextScope::new(scope, ctx);
                                let global = ctx.global(scope);
                                let key = v8::String::new(scope, "_waitForActiveHandles").unwrap();
                                if let Some(func) = global.get(scope, key.into()) {
                                    if func.is_function() {
                                        let func =
                                            v8::Local::<v8::Function>::try_from(func).unwrap();
                                        let recv = v8::undefined(scope).into();
                                        if let Some(result) = func.call(scope, recv, &[]) {
                                            if result.is_promise() {
                                                let promise =
                                                    v8::Local::<v8::Promise>::try_from(result)
                                                        .unwrap();
                                                if promise.state() == v8::PromiseState::Pending {
                                                    execution::set_pending_script_evaluation(
                                                        scope, promise,
                                                    );
                                                }
                                            }
                                        }
                                    }
                                }
                            }

                            // Phase 2: pump event loop for active handles
                            if !pending.is_empty()
                                || execution::has_pending_script_evaluation()
                                || !deferred_queue.lock().unwrap().is_empty()
                            {
                                let scope = &mut v8::HandleScope::new(iso);
                                let ctx = v8::Local::new(scope, &exec_context);
                                let scope = &mut v8::ContextScope::new(scope, ctx);
                                let event_loop_status = run_event_loop(
                                    scope,
                                    &rx,
                                    &pending,
                                    Some(&abort_rx),
                                    Some(&deferred_queue),
                                );

                                if matches!(event_loop_status, EventLoopStatus::Terminated) {
                                    terminated = true;
                                }
                                if let EventLoopStatus::Failed(next_code, next_error) =
                                    event_loop_status
                                {
                                    code = next_code;
                                    error = Some(next_error);
                                }
                            }
                        }

                        if !terminated && mode == 0 && error.is_none() {
                            let scope = &mut v8::HandleScope::new(iso);
                            let ctx = v8::Local::new(scope, &exec_context);
                            let scope = &mut v8::ContextScope::new(scope, ctx);
                            if let Some((next_code, next_error)) =
                                execution::finalize_pending_script_evaluation(scope)
                            {
                                code = next_code;
                                error = next_error;
                            }
                        }

                        // Determine which execution budget (if any) fired. Both the
                        // CPU-time budget and the wall-clock backstop can be armed;
                        // whichever fired first recorded its abort reason. Prefer the
                        // recorded abort reason (first-writer-wins) so the result
                        // attributes termination to the guard that actually fired.
                        let abort_reason = execution_abort_reason(&execution_abort);
                        let wall_clock_timed_out =
                            wall_clock_guard.as_ref().is_some_and(|g| g.timed_out())
                                || matches!(
                                    abort_reason,
                                    Some(ExecutionAbortReason::WallClockTimedOut)
                                );
                        let cpu_budget_exceeded =
                            cpu_budget_guard.as_ref().is_some_and(|g| g.exceeded())
                                || matches!(
                                    abort_reason,
                                    Some(ExecutionAbortReason::CpuBudgetExceeded)
                                );
                        // If both happened to fire, the recorded abort reason is the
                        // authoritative first-fired guard; fall back to wall-clock
                        // only when no CPU-budget reason was recorded.
                        let cpu_budget_exceeded = cpu_budget_exceeded
                            && !matches!(
                                abort_reason,
                                Some(ExecutionAbortReason::WallClockTimedOut)
                            );
                        let wall_clock_timed_out = wall_clock_timed_out && !cpu_budget_exceeded;

                        // Cancel both watchdogs (joins their threads).
                        if let Some(ref mut guard) = cpu_budget_guard {
                            guard.cancel();
                        }
                        drop(cpu_budget_guard);
                        if let Some(ref mut guard) = wall_clock_guard {
                            guard.cancel();
                        }
                        drop(wall_clock_guard);

                        if matches!(abort_reason, Some(ExecutionAbortReason::Terminated)) {
                            terminated = true;
                            code = 1;
                            exports = None;
                            error = None;
                        }
                        if terminated || cpu_budget_exceeded || wall_clock_timed_out {
                            iso.cancel_terminate_execution();
                        }

                        // Send ExecutionResult
                        let result_frame = if cpu_budget_exceeded {
                            if let Some(budget_ms) = cpu_time_limit_ms {
                                let capacity = budget_ms as usize;
                                warn_limit_exhausted(TrackedLimit::V8CpuTimeMs, capacity, capacity);
                            }
                            RuntimeEvent::ExecutionResult {
                                session_id,
                                exit_code: 1,
                                exports: None,
                                error: Some(ExecutionErrorBin {
                                    error_type: "Error".into(),
                                    message: "Script execution exceeded the CPU-time budget \
                                         (limits.jsRuntime.cpuTimeLimitMs)"
                                        .into(),
                                    stack: String::new(),
                                    code: "ERR_SCRIPT_CPU_BUDGET_EXCEEDED".into(),
                                }),
                            }
                        } else if wall_clock_timed_out {
                            if let Some(limit_ms) = wall_clock_limit_ms {
                                let capacity = limit_ms as usize;
                                warn_limit_exhausted(
                                    TrackedLimit::V8WallClockMs,
                                    capacity,
                                    capacity,
                                );
                            }
                            RuntimeEvent::ExecutionResult {
                                session_id,
                                exit_code: 1,
                                exports: None,
                                error: Some(ExecutionErrorBin {
                                    error_type: "Error".into(),
                                    message: "Script execution exceeded the wall-clock limit \
                                         (limits.jsRuntime.wallClockLimitMs)"
                                        .into(),
                                    stack: String::new(),
                                    code: "ERR_SCRIPT_WALL_CLOCK_EXCEEDED".into(),
                                }),
                            }
                        } else if terminated {
                            RuntimeEvent::ExecutionResult {
                                session_id,
                                exit_code: 1,
                                exports: None,
                                error: Some(ExecutionErrorBin {
                                    error_type: "Error".into(),
                                    message: "Execution terminated".into(),
                                    stack: String::new(),
                                    code: String::new(),
                                }),
                            }
                        } else {
                            RuntimeEvent::ExecutionResult {
                                session_id,
                                exit_code: code,
                                exports,
                                error: error.map(|e| ExecutionErrorBin {
                                    error_type: e.error_type,
                                    message: e.message,
                                    stack: e.stack,
                                    code: e.code.unwrap_or_default(),
                                }),
                            }
                        };

                        execution::clear_pending_module_evaluation();
                        execution::clear_pending_script_evaluation();
                        execution::clear_module_state();

                        send_event_with_generation(&event_tx, output_generation, result_frame);
                    }
                    #[cfg(test)]
                    {
                        let _ = (mode, file_path, bridge_code, post_restore_script, user_code);
                    }
                }
                SessionMessage::BridgeResponse(_)
                | SessionMessage::StreamEvent(_)
                | SessionMessage::TerminateExecution => {
                    handle_late_session_message(&event_tx, &session_id, output_generation, msg);
                }
            },
        }
    }

    // Drop V8 resources (only present in non-test mode)
    #[cfg(not(test))]
    {
        *isolate_handle
            .lock()
            .expect("session isolate handle lock poisoned") = None;
        // Reset pending promise-resolver Globals BEFORE the isolate is dropped on
        // thread teardown. run_event_loop can exit early (Shutdown / timeout-abort)
        // with resolvers still registered, so without this the Globals would drop
        // after their isolate — leaking across session create/destroy churn and
        // violating the V8 lifetime contract.
        reset_pending_promises(&mut pending);
        drop(_v8_context.take());
        isolate::drop_isolate(v8_isolate.take());
    }

    // Release concurrency slot
    {
        let (lock, cvar) = &*slot_control;
        let mut count = lock.lock().unwrap();
        *count -= 1;
        cvar.notify_one();
    }
}

/// Sync bridge functions block V8 while the host processes the call
/// (applySync/applySyncPromise). Async bridge functions return a Promise to V8.
struct BridgeFnPartitions {
    sync: Vec<&'static str>,
    async_fns: Vec<&'static str>,
}

pub(crate) fn sync_bridge_fns() -> &'static [&'static str] {
    &bridge_fn_partitions().sync
}

pub(crate) fn async_bridge_fns() -> &'static [&'static str] {
    &bridge_fn_partitions().async_fns
}

fn bridge_fn_partitions() -> &'static BridgeFnPartitions {
    static PARTITIONS: OnceLock<BridgeFnPartitions> = OnceLock::new();
    PARTITIONS.get_or_init(|| BridgeFnPartitions {
        sync: bridge_fns_for(|convention| {
            matches!(
                convention,
                BridgeCallConvention::Sync | BridgeCallConvention::SyncPromise
            )
        }),
        async_fns: bridge_fns_for(|convention| convention == BridgeCallConvention::Async),
    })
}

fn bridge_fns_for(filter: impl Fn(BridgeCallConvention) -> bool) -> Vec<&'static str> {
    bridge_contract()
        .groups
        .iter()
        .filter(|group| filter(group.convention))
        .flat_map(|group| group.names.iter().map(String::as_str))
        .collect()
}

/// Reset every pending promise-resolver `v8::Global` handle held by `pending`.
///
/// `v8::Global` handles MUST be reset/dropped *before* the `v8::Isolate` that
/// created them is torn down. The session reuses a single `PendingPromises`
/// registry across executions and across isolate rebuilds, and `run_event_loop`
/// can exit early (Shutdown at the `SessionCommand::Shutdown` arm, or
/// timeout-abort via the `abort_rx` branch) while resolvers are still
/// registered. On those paths the registry can outlive an isolate. Call this
/// immediately before every isolate drop (rebuild and thread teardown) so the
/// `Global<PromiseResolver>` handles are dropped while their isolate is still
/// alive — preventing both a leak across session create/destroy churn (bounded
/// by `MAX_PENDING_PROMISES`) and a V8 lifetime-contract violation.
#[doc(hidden)]
pub fn reset_pending_promises(pending: &mut crate::bridge::PendingPromises) {
    // Swap in an empty registry and drop the populated one in place. Dropping a
    // `PendingPromises` resets all of its `Global<PromiseResolver>` handles.
    drop(std::mem::take(pending));
}

/// Run the session event loop: dispatch incoming messages to V8.
///
/// Called after script/module execution when there are pending async promises.
/// Polls the session channel for BridgeResponse, StreamEvent, and
/// TerminateExecution messages, dispatching each into V8 with microtask flush.
///
/// When `deferred` is provided, drains queued messages from sync bridge calls
/// before blocking on the channel. This prevents StreamEvent loss when sync
/// bridge calls consume non-BridgeResponse messages from the shared channel.
///
/// When `abort_rx` is provided (timeout is configured), uses `select!` to
/// also monitor the abort channel — if the timeout fires and drops the sender,
/// the abort channel unblocks and terminates execution.
///
/// Returns true if execution completed normally, false if terminated.
#[doc(hidden)]
pub fn run_event_loop(
    scope: &mut v8::HandleScope,
    rx: &Receiver<SessionCommand>,
    pending: &crate::bridge::PendingPromises,
    abort_rx: Option<&crossbeam_channel::Receiver<()>>,
    deferred: Option<&DeferredQueue>,
) -> EventLoopStatus {
    while !pending.is_empty()
        || execution::pending_module_evaluation_needs_wait(scope)
        || execution::pending_script_evaluation_needs_wait(scope)
        || pending_guest_timer_count(scope) > 0
        || pending_guest_immediate_count(scope) > 0
        || deferred
            .map(|dq| !dq.lock().unwrap().is_empty())
            .unwrap_or(false)
    {
        pump_v8_message_loop(scope);

        // Drain deferred messages queued by sync bridge calls before blocking
        if let Some(dq) = deferred {
            let frames: Vec<SessionMessage> = dq.lock().unwrap().drain(..).collect();
            for frame in frames {
                let status = dispatch_event_loop_frame(scope, frame, pending);
                if !matches!(status, EventLoopStatus::Completed) {
                    return status;
                }
            }
            if pending.is_empty()
                && !execution::pending_module_evaluation_needs_wait(scope)
                && !execution::pending_script_evaluation_needs_wait(scope)
                && pending_guest_timer_count(scope) == 0
                && pending_guest_immediate_count(scope) == 0
            {
                break;
            }
        }

        // Flush microtasks before blocking. Run in a loop to drain the full
        // microtask queue -- each checkpoint may resolve Promises that schedule
        // new microtasks (e.g., async function await chains).
        for _ in 0..100 {
            scope.perform_microtask_checkpoint();
            pump_v8_message_loop(scope);
            // Check if new deferred work appeared from microtask processing
            if let Some(dq) = deferred {
                if !dq.lock().unwrap().is_empty() {
                    break; // New bridge work to process
                }
            }
        }

        if pending_guest_immediate_count(scope) > 0 {
            match try_recv_session_command(scope, rx, abort_rx) {
                Ok(Some(cmd)) => {
                    let status = dispatch_session_command(scope, cmd, pending);
                    if !matches!(status, EventLoopStatus::Completed) {
                        return status;
                    }
                }
                Ok(None) => {
                    let status = drain_guest_immediates(scope);
                    if !matches!(status, EventLoopStatus::Completed) {
                        return status;
                    }
                }
                Err(status) => return status,
            }
            scope.perform_microtask_checkpoint();
            pump_v8_message_loop(scope);
        }

        // Re-check exit conditions after microtask flush — the microtask may
        // have resolved all pending promises or registered new handles.
        if pending.is_empty()
            && !execution::pending_module_evaluation_needs_wait(scope)
            && !execution::pending_script_evaluation_needs_wait(scope)
            && pending_guest_timer_count(scope) == 0
            && pending_guest_immediate_count(scope) == 0
            && deferred
                .map(|dq| dq.lock().unwrap().is_empty())
                .unwrap_or(true)
        {
            break;
        }

        // Receive next command with interleaved microtask processing.
        // Instead of blocking indefinitely, use a short timeout so we can
        // periodically flush microtasks (like Node.js's libuv + DrainTasks pattern).
        let cmd = loop {
            if pending_guest_immediate_count(scope) > 0 {
                match try_recv_session_command(scope, rx, abort_rx) {
                    Ok(Some(cmd)) => break cmd,
                    Ok(None) => {
                        let status = drain_guest_immediates(scope);
                        if !matches!(status, EventLoopStatus::Completed) {
                            return status;
                        }
                        scope.perform_microtask_checkpoint();
                        pump_v8_message_loop(scope);
                        continue;
                    }
                    Err(status) => return status,
                }
            }
            let recv_result = if let Some(abort) = abort_rx {
                crossbeam_channel::select! {
                    recv(rx) -> result => result.ok(),
                    recv(abort) -> _ => {
                        scope.terminate_execution();
                        return EventLoopStatus::Terminated;
                    },
                    default(std::time::Duration::from_millis(1)) => None,
                }
            } else {
                rx.recv_timeout(std::time::Duration::from_millis(1)).ok()
            };
            if let Some(cmd) = recv_result {
                break cmd;
            }
            // No command received — flush microtasks and check deferred queue
            scope.perform_microtask_checkpoint();
            pump_v8_message_loop(scope);
            if let Some(dq) = deferred {
                if !dq.lock().unwrap().is_empty() {
                    // New deferred work appeared — drain it in the outer loop
                    let frames: Vec<SessionMessage> = dq.lock().unwrap().drain(..).collect();
                    for frame in frames {
                        let status = dispatch_event_loop_frame(scope, frame, pending);
                        if !matches!(status, EventLoopStatus::Completed) {
                            return status;
                        }
                    }
                }
            }
            // Check if we should exit
            if pending.is_empty()
                && !execution::pending_module_evaluation_needs_wait(scope)
                && !execution::pending_script_evaluation_needs_wait(scope)
                && pending_guest_timer_count(scope) == 0
                && pending_guest_immediate_count(scope) == 0
                && deferred
                    .map(|dq| dq.lock().unwrap().is_empty())
                    .unwrap_or(true)
            {
                return EventLoopStatus::Completed;
            }
        };

        let status = dispatch_session_command(scope, cmd, pending);
        if !matches!(status, EventLoopStatus::Completed) {
            return status;
        }
    }
    EventLoopStatus::Completed
}

fn try_recv_session_command(
    scope: &mut v8::HandleScope,
    rx: &Receiver<SessionCommand>,
    abort_rx: Option<&crossbeam_channel::Receiver<()>>,
) -> Result<Option<SessionCommand>, EventLoopStatus> {
    if let Some(abort) = abort_rx {
        crossbeam_channel::select! {
            recv(abort) -> _ => {
                scope.terminate_execution();
                Err(EventLoopStatus::Terminated)
            },
            recv(rx) -> result => Ok(result.ok()),
            default => Ok(None),
        }
    } else {
        match rx.try_recv() {
            Ok(cmd) => Ok(Some(cmd)),
            Err(crossbeam_channel::TryRecvError::Empty) => Ok(None),
            Err(crossbeam_channel::TryRecvError::Disconnected) => Ok(None),
        }
    }
}

fn dispatch_session_command(
    scope: &mut v8::HandleScope,
    cmd: SessionCommand,
    pending: &crate::bridge::PendingPromises,
) -> EventLoopStatus {
    match cmd {
        SessionCommand::Message(frame) => dispatch_event_loop_frame(scope, frame, pending),
        SessionCommand::SetModuleReader(reader) => {
            execution::install_session_guest_reader(Some(reader));
            EventLoopStatus::Completed
        }
        SessionCommand::Shutdown => EventLoopStatus::Terminated,
    }
}

fn pending_guest_timer_count(scope: &mut v8::HandleScope) -> usize {
    let tc = &mut v8::TryCatch::new(scope);
    let context = tc.get_current_context();
    let global = context.global(tc);
    let key = match v8::String::new(tc, "_getPendingTimerCount") {
        Some(key) => key,
        None => return 0,
    };
    let Some(func_value) = global.get(tc, key.into()) else {
        return 0;
    };
    let Ok(func) = v8::Local::<v8::Function>::try_from(func_value) else {
        return 0;
    };
    let Some(result) = func.call(tc, global.into(), &[]) else {
        return 0;
    };

    result
        .integer_value(tc)
        .and_then(|count| usize::try_from(count).ok())
        .unwrap_or(0)
}

fn pending_guest_immediate_count(scope: &mut v8::HandleScope) -> usize {
    let tc = &mut v8::TryCatch::new(scope);
    let context = tc.get_current_context();
    let global = context.global(tc);
    let key = match v8::String::new(tc, "_getPendingImmediateCount") {
        Some(key) => key,
        None => return 0,
    };
    let Some(func_value) = global.get(tc, key.into()) else {
        return 0;
    };
    let Ok(func) = v8::Local::<v8::Function>::try_from(func_value) else {
        return 0;
    };
    let Some(result) = func.call(tc, global.into(), &[]) else {
        return 0;
    };

    result
        .integer_value(tc)
        .and_then(|count| usize::try_from(count).ok())
        .unwrap_or(0)
}

fn drain_guest_immediates(scope: &mut v8::HandleScope) -> EventLoopStatus {
    let tc = &mut v8::TryCatch::new(scope);
    let context = tc.get_current_context();
    let global = context.global(tc);
    let key = match v8::String::new(tc, "_drainImmediates") {
        Some(key) => key,
        None => return EventLoopStatus::Completed,
    };
    let Some(func_value) = global.get(tc, key.into()) else {
        return EventLoopStatus::Completed;
    };
    let Ok(func) = v8::Local::<v8::Function>::try_from(func_value) else {
        return EventLoopStatus::Completed;
    };
    let _ = func.call(tc, global.into(), &[]);
    tc.perform_microtask_checkpoint();
    if let Some(exception) = tc.exception() {
        let (code, err) = execution::exception_to_result(tc, exception);
        return EventLoopStatus::Failed(code, err);
    }
    if let Some(err) = execution::take_unhandled_promise_rejection(tc) {
        return EventLoopStatus::Failed(1, err);
    }
    EventLoopStatus::Completed
}

fn pump_v8_message_loop(scope: &mut v8::HandleScope) {
    let platform = v8::V8::get_current_platform();
    while v8::Platform::pump_message_loop(&platform, scope, false) {
        scope.perform_microtask_checkpoint();
    }
}

/// Dispatch a single session message within the event loop.
/// Returns the event-loop status after handling the frame.
#[derive(Debug)]
#[doc(hidden)]
pub enum EventLoopStatus {
    Completed,
    Terminated,
    Failed(i32, ExecutionError),
}

fn dispatch_event_loop_frame(
    scope: &mut v8::HandleScope,
    frame: SessionMessage,
    pending: &crate::bridge::PendingPromises,
) -> EventLoopStatus {
    match frame {
        SessionMessage::BridgeResponse(BridgeResponse {
            call_id,
            status,
            payload,
        }) => {
            let (result, error) = if status == 1 {
                (None, Some(String::from_utf8_lossy(&payload).to_string()))
            } else if status == 2 || !payload.is_empty() {
                // status=0: V8-serialized, status=2: raw binary (Uint8Array)
                (Some(payload), None)
            } else {
                (None, None)
            };
            let _ = crate::bridge::resolve_pending_promise(
                scope, pending, call_id, status, result, error,
            );
            // Microtasks already flushed in resolve_pending_promise
            EventLoopStatus::Completed
        }
        SessionMessage::StreamEvent(StreamEvent {
            event_type,
            payload,
        }) => {
            let tc = &mut v8::TryCatch::new(scope);
            crate::stream::dispatch_stream_event(tc, &event_type, &payload);
            tc.perform_microtask_checkpoint();
            if let Some(exception) = tc.exception() {
                let (code, err) = execution::exception_to_result(tc, exception);
                return EventLoopStatus::Failed(code, err);
            }
            if let Some(err) = execution::take_unhandled_promise_rejection(tc) {
                return EventLoopStatus::Failed(1, err);
            }
            EventLoopStatus::Completed
        }
        SessionMessage::TerminateExecution => {
            scope.terminate_execution();
            EventLoopStatus::Terminated
        }
        _ => {
            // Ignore other messages during event loop
            EventLoopStatus::Completed
        }
    }
}

/// ResponseReceiver that receives typed session messages directly from the session channel.
///
/// Only returns BridgeResponse frames from recv_response(). Non-BridgeResponse
/// messages (StreamEvent, TerminateExecution) consumed during sync bridge calls
/// are queued in the deferred queue for later processing by the event loop.
///
/// When `abort_rx` is set (timeout configured), uses `select!` to also monitor
/// the abort channel. If the timeout fires, the abort sender is dropped, which
/// unblocks the select and returns a timeout error.
pub(crate) struct ChannelResponseReceiver {
    rx: Receiver<SessionCommand>,
    abort_rx: Option<crossbeam_channel::Receiver<()>>,
    deferred: DeferredQueue,
}

impl ChannelResponseReceiver {
    #[allow(dead_code)]
    pub(crate) fn new(rx: Receiver<SessionCommand>, deferred: DeferredQueue) -> Self {
        ChannelResponseReceiver {
            rx,
            abort_rx: None,
            deferred,
        }
    }

    pub(crate) fn with_abort(
        rx: Receiver<SessionCommand>,
        abort_rx: crossbeam_channel::Receiver<()>,
        deferred: DeferredQueue,
    ) -> Self {
        ChannelResponseReceiver {
            rx,
            abort_rx: Some(abort_rx),
            deferred,
        }
    }
}

impl crate::host_call::BridgeResponseReceiver for ChannelResponseReceiver {
    fn recv_response(&self, expected_call_id: u64) -> Result<BridgeResponse, String> {
        loop {
            // Wait for next command, with optional abort monitoring
            let cmd = if let Some(ref abort) = self.abort_rx {
                crossbeam_channel::select! {
                    recv(self.rx) -> result => match result {
                        Ok(cmd) => cmd,
                        Err(_) => return Err("channel closed".into()),
                    },
                    recv(abort) -> _ => {
                        return Err("execution aborted".into());
                    },
                }
            } else {
                match self.rx.recv() {
                    Ok(cmd) => cmd,
                    Err(_) => return Err("channel closed".into()),
                }
            };

            match cmd {
                SessionCommand::Message(frame) => {
                    if let SessionMessage::BridgeResponse(response) = &frame {
                        let call_id = response.call_id;
                        if call_id == expected_call_id {
                            crate::host_call::record_sync_bridge_response_channel_received(call_id);
                            return Ok(response.clone());
                        }
                        push_deferred_sync_message(&self.deferred, frame)?;
                        continue;
                    }
                    // Queue non-BridgeResponse for later event loop processing
                    push_deferred_sync_message(&self.deferred, frame)?;
                }
                SessionCommand::SetModuleReader(reader) => {
                    execution::install_session_guest_reader(Some(reader));
                }
                SessionCommand::Shutdown => return Err("session shutdown".into()),
            }
        }
    }
}

fn push_deferred_sync_message(
    deferred: &DeferredQueue,
    frame: SessionMessage,
) -> Result<(), String> {
    let mut queue = deferred.lock().unwrap();
    if queue.len() >= MAX_DEFERRED_SYNC_MESSAGES {
        return Err(format!(
            "sync bridge deferred message queue exceeded limit of {MAX_DEFERRED_SYNC_MESSAGES}"
        ));
    }
    queue.push_back(frame);
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::{HashMap, HashSet};

    /// Helper to create a SessionManager for tests
    fn test_manager(max: usize) -> SessionManager {
        test_manager_with_events(max).0
    }

    fn test_manager_with_events(max: usize) -> (SessionManager, Receiver<RuntimeEventEnvelope>) {
        let (tx, _rx) = crossbeam_channel::unbounded();
        let router: CallIdRouter = Arc::new(Mutex::new(HashMap::new()));
        let snap_cache = Arc::new(SnapshotCache::new(4));
        let manager = SessionManager::new(max, tx, router, snap_cache);
        (manager, _rx)
    }

    #[test]
    fn zero_cpu_time_limit_is_normalized_to_no_timeout() {
        assert_eq!(normalize_cpu_time_limit_ms(None), None);
        assert_eq!(normalize_cpu_time_limit_ms(Some(0)), None);
        assert_eq!(normalize_cpu_time_limit_ms(Some(1)), Some(1));
    }

    fn expect_late_message_warning(
        rx: &Receiver<RuntimeEventEnvelope>,
        session_id: &str,
        error_code: &str,
        detail_fragment: &str,
    ) {
        let event = rx
            .recv_timeout(std::time::Duration::from_millis(200))
            .expect("late-message warning");
        match event.event {
            RuntimeEvent::Log {
                session_id: observed_session_id,
                channel,
                message,
            } => {
                assert_eq!(observed_session_id, session_id);
                assert_eq!(channel, 1, "late warnings should use stderr channel");
                assert!(
                    message.contains(error_code),
                    "warning should contain error code {error_code}, got {message}"
                );
                assert!(
                    message.contains(detail_fragment),
                    "warning should mention {detail_fragment}, got {message}"
                );
            }
            other => panic!("expected late-message warning log, got {other:?}"),
        }
    }

    #[test]
    fn bridge_contract_function_partitions_cover_contract() {
        let contract = bridge_contract();

        let expected_sync = contract
            .groups
            .iter()
            .filter(|group| {
                matches!(
                    group.convention,
                    BridgeCallConvention::Sync | BridgeCallConvention::SyncPromise
                )
            })
            .flat_map(|group| group.names.iter().map(String::as_str))
            .collect::<HashSet<_>>();
        let expected_async = contract
            .groups
            .iter()
            .filter(|group| group.convention == BridgeCallConvention::Async)
            .flat_map(|group| group.names.iter().map(String::as_str))
            .collect::<HashSet<_>>();

        let sync_names = sync_bridge_fns();
        let async_names = async_bridge_fns();
        let registered_sync = sync_names.iter().copied().collect::<HashSet<_>>();
        let registered_async = async_names.iter().copied().collect::<HashSet<_>>();

        assert_eq!(
            registered_sync, expected_sync,
            "sync bridge function partition drifted from crates/bridge/bridge-contract.json"
        );
        assert_eq!(
            registered_async, expected_async,
            "async bridge function partition drifted from crates/bridge/bridge-contract.json"
        );
        assert!(
            registered_sync.is_disjoint(&registered_async),
            "sync and async bridge function partitions must not overlap"
        );
    }

    #[test]
    fn session_management() {
        // Consolidated test to avoid V8 inter-test SIGSEGV issues.
        // Covers: lifecycle and concurrency queuing.

        // --- Part 1: Single session create/destroy ---
        {
            let mut mgr = test_manager(4);

            mgr.create_session("session-aaa".into(), None, None, None)
                .expect("create session A");
            assert_eq!(mgr.session_count(), 1);

            // Wait for thread to acquire slot and create isolate
            std::thread::sleep(std::time::Duration::from_millis(200));

            // Destroy session A
            mgr.destroy_session("session-aaa")
                .expect("destroy session A");
            assert_eq!(mgr.session_count(), 0);
        }

        // --- Part 2: Multiple sessions ---
        {
            let mut mgr = test_manager(4);

            mgr.create_session("session-bbb".into(), None, None, None)
                .expect("create session B");
            mgr.create_session("session-ccc".into(), Some(16), None, None)
                .expect("create session C");
            assert_eq!(mgr.session_count(), 2);

            std::thread::sleep(std::time::Duration::from_millis(200));

            // Duplicate session ID is rejected
            let err = mgr.create_session("session-bbb".into(), None, None, None);
            assert!(err.is_err());
            assert!(err.unwrap_err().contains("already exists"));

            // Sending to a missing session still fails.
            let err = mgr.send_to_session("missing", SessionMessage::TerminateExecution);
            assert!(err.is_err());
            assert!(err.unwrap_err().contains("does not exist"));

            // Destroy non-existent session
            let err = mgr.destroy_session("no-such-session");
            assert!(err.is_err());
            assert!(err.unwrap_err().contains("does not exist"));

            mgr.destroy_sessions(["session-bbb".into(), "session-ccc".into()]);
            assert_eq!(mgr.session_count(), 0);
        }

        // --- Part 3: Max concurrency queuing ---
        {
            let mut mgr = test_manager(2);

            mgr.create_session("s1".into(), None, None, None)
                .expect("create s1");
            mgr.create_session("s2".into(), None, None, None)
                .expect("create s2");
            mgr.create_session("s3".into(), None, None, None)
                .expect("create s3");

            // Allow threads to acquire slots
            std::thread::sleep(std::time::Duration::from_millis(300));

            // Only 2 slots active (s3 is queued)
            assert_eq!(mgr.active_slot_count(), 2);
            assert_eq!(mgr.session_count(), 3);

            // Destroy s1 — releases slot, s3 acquires it
            mgr.destroy_session("s1").expect("destroy s1");
            std::thread::sleep(std::time::Duration::from_millis(300));
            assert_eq!(mgr.active_slot_count(), 2);
            assert_eq!(mgr.session_count(), 2);

            // Destroy remaining
            mgr.destroy_sessions(["s2".into(), "s3".into()]);
            std::thread::sleep(std::time::Duration::from_millis(100));
            assert_eq!(mgr.session_count(), 0);
            assert_eq!(mgr.active_slot_count(), 0);
        }
    }

    #[test]
    fn detach_session_clears_call_id_routes_for_session() {
        let mut mgr = test_manager(1);
        mgr.create_session_with_output_generation(
            "session-route".into(),
            None,
            None,
            None,
            Some(7),
            None,
        )
        .expect("create session");
        mgr.call_id_router()
            .lock()
            .expect("call_id router")
            .insert(42, "session-route".into());

        assert!(
            mgr.detach_session_if_output_generation("session-route", 7)
                .expect("detach session"),
            "matching output generation should detach session"
        );
        assert!(
            mgr.call_id_router()
                .lock()
                .expect("call_id router")
                .get(&42)
                .is_none(),
            "detach should clear stale bridge call routes for the session"
        );
    }

    #[test]
    fn begin_destroy_session_removes_entry_before_finish() {
        let mut mgr = test_manager(1);
        mgr.create_session("two-phase".into(), None, None, None)
            .expect("create session");

        let first_shutdown = mgr
            .begin_destroy_session("two-phase")
            .expect("begin destroy session");
        assert_eq!(
            mgr.session_count(),
            0,
            "entry should be removed before the shutdown is finished"
        );

        // A same-id create during the unfinished shutdown window must succeed
        // because the entry was removed up front.
        mgr.create_session("two-phase".into(), None, None, None)
            .expect("re-create session while first shutdown is unfinished");

        let second_shutdown = mgr
            .begin_destroy_session("two-phase")
            .expect("begin destroy re-created session");
        first_shutdown.finish();
        second_shutdown.finish();
        assert_eq!(mgr.session_count(), 0);
    }

    #[test]
    fn session_shutdown_finish_clears_late_call_routes() {
        let mut mgr = test_manager(1);
        mgr.create_session("late-route".into(), None, None, None)
            .expect("create session");

        let shutdown = mgr
            .begin_destroy_session("late-route")
            .expect("begin destroy session");
        // Simulate a route the session thread registered between the pre-join
        // route clear and thread exit.
        mgr.call_id_router()
            .lock()
            .expect("call_id router")
            .insert(42, "late-route".into());

        shutdown.finish();
        assert!(
            mgr.call_id_router()
                .lock()
                .expect("call_id router")
                .get(&42)
                .is_none(),
            "finish should clear call routes registered during shutdown"
        );
    }

    #[test]
    fn channel_response_receiver_filters_bridge_response() {
        use crate::host_call::BridgeResponseReceiver;

        // Sync bridge call interleaved with StreamEvent does not drop the StreamEvent
        let (tx, rx) = crossbeam_channel::bounded(10);
        let deferred = new_deferred_queue();
        let receiver = ChannelResponseReceiver::new(rx, Arc::clone(&deferred));

        // Send: StreamEvent, TerminateExecution, then BridgeResponse
        tx.send(SessionCommand::Message(SessionMessage::StreamEvent(
            StreamEvent {
                event_type: "child_stdout".into(),
                payload: vec![0x01, 0x02],
            },
        )))
        .unwrap();
        tx.send(SessionCommand::Message(SessionMessage::TerminateExecution))
            .unwrap();
        tx.send(SessionCommand::Message(SessionMessage::BridgeResponse(
            BridgeResponse {
                call_id: 1,
                status: 0,
                payload: vec![0xAB],
            },
        )))
        .unwrap();

        // recv_response should skip StreamEvent and TerminateExecution, return BridgeResponse
        let frame = receiver.recv_response(1).unwrap();
        assert!(
            frame.call_id == 1,
            "expected BridgeResponse with call_id=1, got {:?}",
            frame
        );

        // Deferred queue should contain the StreamEvent and TerminateExecution
        let dq = deferred.lock().unwrap();
        assert_eq!(dq.len(), 2, "expected 2 deferred messages");
        assert!(
            matches!(&dq[0], SessionMessage::StreamEvent(StreamEvent { event_type, .. }) if event_type == "child_stdout"),
            "first deferred should be StreamEvent"
        );
        assert!(
            matches!(&dq[1], SessionMessage::TerminateExecution),
            "second deferred should be TerminateExecution"
        );
    }

    #[test]
    fn channel_response_receiver_rejects_deferred_queue_overflow() {
        use crate::host_call::BridgeResponseReceiver;

        let (tx, rx) = crossbeam_channel::bounded(MAX_DEFERRED_SYNC_MESSAGES + 1);
        let deferred = new_deferred_queue();
        let receiver = ChannelResponseReceiver::new(rx, Arc::clone(&deferred));

        for index in 0..=MAX_DEFERRED_SYNC_MESSAGES {
            tx.send(SessionCommand::Message(SessionMessage::StreamEvent(
                StreamEvent {
                    event_type: format!("child_stdout_{index}"),
                    payload: Vec::new(),
                },
            )))
            .unwrap();
        }

        let error = receiver
            .recv_response(1)
            .expect_err("deferred queue overflow should reject sync bridge wait");
        assert!(error.contains("deferred message queue exceeded limit"));
        assert_eq!(deferred.lock().unwrap().len(), MAX_DEFERRED_SYNC_MESSAGES);
    }

    #[test]
    fn pre_slot_deferred_command_overflow_is_bounded_and_logged() {
        let (event_tx, event_rx) = crossbeam_channel::unbounded();
        let mut deferred_commands = VecDeque::new();

        for _ in 0..MAX_DEFERRED_SESSION_COMMANDS {
            assert!(defer_session_command_before_slot(
                &mut deferred_commands,
                &event_tx,
                "queued-session",
                Some(3),
                SessionCommand::Message(SessionMessage::TerminateExecution),
            ));
        }

        assert!(!defer_session_command_before_slot(
            &mut deferred_commands,
            &event_tx,
            "queued-session",
            Some(3),
            SessionCommand::Message(SessionMessage::TerminateExecution),
        ));
        assert_eq!(deferred_commands.len(), MAX_DEFERRED_SESSION_COMMANDS);

        let warning = event_rx.recv().expect("overflow warning");
        assert_eq!(warning.output_generation, Some(3));
        match warning.event {
            RuntimeEvent::Log {
                session_id,
                channel,
                message,
            } => {
                assert_eq!(session_id, "queued-session");
                assert_eq!(channel, 1);
                assert!(message.contains(DEFERRED_COMMAND_LIMIT_ERROR_CODE));
            }
            other => panic!("expected overflow warning log, got {other:?}"),
        }
    }

    #[test]
    fn late_terminate_execution_is_logged_instead_of_silently_dropped() {
        let (mut mgr, rx) = test_manager_with_events(1);
        mgr.create_session("late-terminate".into(), None, None, None)
            .expect("create session");

        mgr.send_to_session("late-terminate", SessionMessage::TerminateExecution)
            .expect("send late terminate");

        expect_late_message_warning(
            &rx,
            "late-terminate",
            LATE_TERMINATE_EXECUTION_ERROR_CODE,
            "TerminateExecution",
        );

        mgr.destroy_session("late-terminate")
            .expect("destroy session");
    }

    #[test]
    fn channel_response_receiver_abort_unblocks_waiting_sync_call() {
        use crate::host_call::BridgeResponseReceiver;

        let (_tx, rx) = crossbeam_channel::bounded(1);
        let deferred = new_deferred_queue();
        let execution_abort = new_execution_abort();
        let (_active_abort, abort_rx) = ActiveExecutionAbort::arm(&execution_abort);
        let receiver = ChannelResponseReceiver::with_abort(rx, abort_rx, deferred);

        let (result_tx, result_rx) = std::sync::mpsc::channel();
        let join_handle = std::thread::spawn(move || {
            let _ = result_tx.send(receiver.recv_response(1));
        });

        std::thread::sleep(std::time::Duration::from_millis(25));
        signal_execution_abort(&execution_abort, ExecutionAbortReason::Terminated);

        let result = result_rx
            .recv_timeout(std::time::Duration::from_secs(1))
            .expect("abort should unblock the waiting receiver");
        assert_eq!(
            result.expect_err("abort should not yield a bridge response"),
            "execution aborted"
        );

        join_handle
            .join()
            .expect("receiver thread should exit cleanly");
    }

    #[test]
    fn late_bridge_response_is_logged_instead_of_silently_dropped() {
        let (mut mgr, rx) = test_manager_with_events(1);
        mgr.create_session("late-bridge".into(), None, None, None)
            .expect("create session");

        mgr.send_to_session(
            "late-bridge",
            SessionMessage::BridgeResponse(BridgeResponse {
                call_id: 41,
                status: 0,
                payload: vec![0xAA, 0xBB],
            }),
        )
        .expect("send late bridge response");

        expect_late_message_warning(
            &rx,
            "late-bridge",
            LATE_BRIDGE_RESPONSE_ERROR_CODE,
            "BridgeResponse",
        );

        mgr.destroy_session("late-bridge").expect("destroy session");
    }

    /// Regression test for the pending-promise-resolver leak / V8 lifetime-contract
    /// violation: when `run_event_loop` exits early (Shutdown or timeout-abort) the
    /// `PendingPromises` registry can still hold `Global<PromiseResolver>` handles,
    /// and the session-thread teardown must reset them *before* dropping the isolate.
    ///
    /// This drives the real cleanup seam (`reset_pending_promises`) used on every
    /// isolate-drop path. It populates the registry with live resolver Globals (as a
    /// terminated execution would leave behind), runs the cleanup while the isolate
    /// is still alive, and asserts the registry is empty (every Global dropped).
    ///
    /// Fast + bounded (a handful of resolvers, then the safeguard fires) — it asserts
    /// the cleanup happens, it does not saturate `MAX_PENDING_PROMISES`.
    #[test]
    fn reset_pending_promises_drops_resolver_globals_before_isolate_teardown() {
        use crate::bridge::{register_async_bridge_fns, PendingPromises};
        use crate::host_call::BridgeCallContext;
        use crate::isolate;
        use std::process::Command;

        // V8 isolates must be created in an isolated process: doing it inline in a
        // parallel `cargo test` thread races the process-global V8 platform and
        // segfaults. Re-exec this one test as a subprocess (matching the crate's
        // bridge_v8_hardening_* / vm_context_registry convention).
        const SUBPROCESS_ENV: &str = "AGENTOS_V8_RESET_PENDING_PROMISES_SUBPROCESS";
        if std::env::var_os(SUBPROCESS_ENV).is_none() {
            let output = Command::new(std::env::current_exe().expect("current test binary"))
                .arg("session::tests::reset_pending_promises_drops_resolver_globals_before_isolate_teardown")
                .arg("--exact")
                .arg("--nocapture")
                .env(SUBPROCESS_ENV, "1")
                .output()
                .expect("spawn reset-pending-promises subprocess");
            assert!(
                output.status.success(),
                "reset-pending-promises subprocess failed with status {:?}\nstdout:\n{}\nstderr:\n{}",
                output.status.code(),
                String::from_utf8_lossy(&output.stdout),
                String::from_utf8_lossy(&output.stderr)
            );
            return;
        }

        isolate::init_v8_platform();

        let mut v8_isolate = isolate::create_isolate(None);
        let context = isolate::create_context(&mut v8_isolate);
        let scope = &mut v8::HandleScope::new(&mut v8_isolate);
        let context = v8::Local::new(scope, &context);
        let scope = &mut v8::ContextScope::new(scope, context);

        let bridge_ctx = BridgeCallContext::new(
            Box::new(std::io::sink()),
            Box::new(std::io::empty()),
            String::from("reset-pending-test"),
        );
        let mut pending = PendingPromises::new();

        // Each `_asyncFn(i)` call synchronously registers a pending promise
        // resolver Global in `pending` and returns an unresolved Promise —
        // exactly what remains registered when the event loop exits early on
        // Shutdown / timeout-abort.
        const REGISTERED: usize = 8;
        let _async_fns = register_async_bridge_fns(
            scope,
            &bridge_ctx as *const BridgeCallContext,
            &pending as *const PendingPromises,
            &["_asyncFn"],
        );
        let source = format!("for (let i = 0; i < {REGISTERED}; i++) {{ _asyncFn(i); }}");
        {
            let tc = &mut v8::TryCatch::new(scope);
            let code = v8::String::new(tc, &source).unwrap();
            let script = v8::Script::compile(tc, code, None).unwrap();
            assert!(
                script.run(tc).is_some(),
                "async bridge calls should register resolvers, not throw"
            );
            assert!(!tc.has_caught(), "async bridge calls should not throw");
        }
        assert_eq!(
            pending.len(),
            REGISTERED,
            "each _asyncFn call must register a pending resolver Global"
        );

        // The cleanup invoked on every session-thread isolate-drop path. It must
        // empty the registry (resetting every Global<PromiseResolver>) while the
        // isolate is still alive.
        reset_pending_promises(&mut pending);

        assert_eq!(
            pending.len(),
            0,
            "reset_pending_promises must drop all pending resolver Globals before isolate teardown"
        );

        // Isolate is still alive here: the Globals were reset above, so dropping
        // the scope/isolate below honors the V8 lifetime contract.
    }
}