diff --git a/crates/agentos-sidecar/src/acp/turn.rs b/crates/agentos-sidecar/src/acp/turn.rs index 89476c093a..fc990de3a6 100644 --- a/crates/agentos-sidecar/src/acp/turn.rs +++ b/crates/agentos-sidecar/src/acp/turn.rs @@ -1017,20 +1017,24 @@ impl DurableUpdateSink { } // Tool, plan, mode, and other durable updates may be interleaved with - // message deltas. Keep their native order inside the in-progress - // completion buffer instead of treating them as a message boundary. - if self.buffered_kind.is_some() { - self.buffered_bytes = checked_acp_bytes( - &self.user_session_id, - self.buffered_bytes, - update_bytes, - self.limits.max_completed_message_bytes, - "limits.acp.maxCompletedMessageBytes", - )?; - self.buffered.push(update); - } else { - self.persist(ctx, events, vec![update]).await?; - } + // message deltas. Commit the in-progress text run first, then this + // update, so it reaches the host the moment the adapter produces it. + // + // Holding it in the completion buffer instead would keep it invisible + // until the next message boundary: an agent that streams any text + // before calling a tool (Pi prints a banner chunk on some turns) arms + // the buffer, and every later `tool_call` / `tool_call_update` is then + // withheld until the *post-tool* message arrives — so a + // `tool_call_update { in_progress }` for a `sleep 60` reaches the host + // only when the command finishes, and a caller waiting on that + // boundary before cancelling never gets a live turn to cancel. + // + // The durable sequence is unchanged: `coalesce_completed_message` + // already ends the text run it is building at any non-text update, so + // flushing here emits the same events in the same order, just in more + // (smaller) batches. + self.flush(ctx, events).await?; + self.persist(ctx, events, vec![update]).await?; Ok(true) } diff --git a/crates/agentos-sidecar/tests/acp_live_updates.rs b/crates/agentos-sidecar/tests/acp_live_updates.rs new file mode 100644 index 0000000000..173d783c77 --- /dev/null +++ b/crates/agentos-sidecar/tests/acp_live_updates.rs @@ -0,0 +1,500 @@ +//! Regression guard: durable `session/update`s must stream live mid-turn even +//! when the agent emitted a message chunk first. +//! +//! Original bug: `DurableUpdateSink::handle_notification` buffered every +//! non-message update (`tool_call`, `tool_call_update`, `plan`, ...) whenever a +//! message chunk had already opened a completion buffer, and only committed the +//! buffer at the next message boundary. An agent that streams any text before +//! calling a tool therefore withheld the whole durable stream until the +//! *post-tool* message arrived, so a `tool_call_update { in_progress }` for a +//! long-running tool reached the host only when the tool finished — and a caller +//! that waits for that boundary before cancelling never saw a live turn to +//! cancel. +//! +//! The adapter here reproduces exactly that ordering: one `agent_message_chunk`, +//! then the tool updates, then a hold, then the prompt response. With live +//! delivery the tool update reaches the event sink during the hold; with the bug +//! it arrives only when the prompt resolves. +//! +//! A separate test file keeps this guard standalone (see the note in +//! `acp_request_timeout.rs`); it installs its own `EventSinkTransport`, which +//! an in-process `NativeSidecar` otherwise does not have. + +#[path = "support/bridge.rs"] +mod bridge_support; + +use std::collections::HashMap; +use std::fs; +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use agentos_native_sidecar::protocol::EventPayload; +use agentos_native_sidecar::wire::{ + event_frame_to_compat, AuthenticateRequest, ConfigureVmRequest, ConnectionOwnership, + CreateVmRequest, ExtEnvelope, GuestRuntimeKind, OpenSessionRequest, OwnershipScope, + PackageDescriptor, RequestFrame, RequestPayload, ResponsePayload, SessionOwnership, + SidecarPlacement, SidecarPlacementShared, VmOwnership, +}; +use agentos_native_sidecar::{ + EventSinkTransport, NativeSidecar, NativeSidecarConfig, SidecarError, +}; +use agentos_protocol::generated::v1::{ + AcpDurableEvent, AcpEvent, AcpOpenSessionRequest, AcpPromptRequest, AcpRequest, AcpResponse, +}; +use agentos_protocol::ACP_EXTENSION_NAMESPACE; +use agentos_vm_config as vm_config; +use bridge_support::RecordingBridge; + +/// How long the adapter holds the prompt open after producing the tool updates. +const ADAPTER_HOLD_MS: u64 = 2_000; + +#[test] +fn tool_updates_stream_live_after_a_leading_message_chunk() { + assert_node_available(); + let mut sidecar = new_sidecar("live-updates"); + let sink = Arc::new(RecordingEventSink::default()); + sidecar.set_event_transport(sink.clone()); + + let connection_id = authenticate(&mut sidecar); + let session_id = open_session(&mut sidecar, &connection_id); + let cwd = temp_dir("live-updates-cwd"); + fs::write(cwd.join(ADAPTER_FILE), adapter_script()).expect("write adapter script"); + let vm_id = create_vm(&mut sidecar, &connection_id, &session_id, &cwd); + + let opened = dispatch_acp( + &mut sidecar, + 4, + &connection_id, + &session_id, + &vm_id, + AcpRequest::AcpOpenSessionRequest(AcpOpenSessionRequest { + session_id: Some(String::from("live-session")), + agent: String::from("pi"), + cwd: Some(String::from("/home/agentos")), + additional_directories: None, + env: None, + mcp_servers: None, + permission_policy: None, + skip_os_instructions: Some(true), + additional_instructions: None, + }), + ); + assert!( + matches!(opened, AcpResponse::AcpOpenSessionResponse(_)), + "expected the mock ACP adapter to open a session, got: {opened:?}" + ); + + sink.reset(); + let started = Instant::now(); + let response = dispatch_acp( + &mut sidecar, + 6, + &connection_id, + &session_id, + &vm_id, + AcpRequest::AcpPromptRequest(AcpPromptRequest { + session_id: Some(String::from("live-session")), + idempotency_key: None, + content: String::from(r#"[{"type":"text","text":"run the tool"}]"#), + }), + ); + let resolved = started.elapsed(); + assert!( + matches!(response, AcpResponse::AcpPromptResponse(_)), + "expected the prompt to complete, got: {response:?}" + ); + assert!( + resolved >= Duration::from_millis(ADAPTER_HOLD_MS), + "the adapter must really hold the turn open, otherwise there is no \ + mid-turn window to observe; turn took {resolved:?}" + ); + + let in_progress = sink + .first_durable_update_containing("\"in_progress\"") + .expect( + "the tool_call_update { in_progress } must reach the host at all — no durable \ + session update carrying it was emitted", + ); + assert!( + in_progress < Duration::from_millis(ADAPTER_HOLD_MS / 2), + "BUG: tool_call_update {{ in_progress }} arrived after {in_progress:?}, i.e. it was \ + held in the message-completion buffer until the turn ended (turn: {resolved:?}) \ + instead of streaming when the adapter produced it", + ); +} + +#[derive(Default)] +struct RecordingEventSink { + started: Mutex>, + events: Mutex>, +} + +impl RecordingEventSink { + fn reset(&self) { + *self.started.lock().expect("sink clock") = Some(Instant::now()); + self.events.lock().expect("sink events").clear(); + } + + /// Elapsed time from `reset` to the first durable session update whose JSON + /// contains `needle`. + fn first_durable_update_containing(&self, needle: &str) -> Option { + self.events + .lock() + .expect("sink events") + .iter() + .find_map(|(at, event)| match event { + AcpEvent::AcpDurableSessionEvent(durable) => match &durable.event { + AcpDurableEvent::AcpDurableSessionUpdate(update) + if update.update.contains(needle) => + { + Some(*at) + } + _ => None, + }, + _ => None, + }) + } +} + +impl EventSinkTransport for RecordingEventSink { + fn emit_event( + &self, + event: agentos_native_sidecar::wire::EventFrame, + ) -> Result<(), SidecarError> { + let at = self + .started + .lock() + .expect("sink clock") + .map(|started| started.elapsed()) + .unwrap_or_default(); + let frame = event_frame_to_compat(event) + .map_err(|error| SidecarError::InvalidState(error.to_string()))?; + if let EventPayload::Ext(ExtEnvelope { namespace, payload }) = frame.payload { + if namespace == ACP_EXTENSION_NAMESPACE { + if let Ok(event) = serde_bare::from_slice::(&payload) { + self.events.lock().expect("sink events").push((at, event)); + } + } + } + Ok(()) + } +} + +const ADAPTER_FILE: &str = "live-updates-adapter.mjs"; + +/// Adapter that handshakes normally, then on `session/prompt` streams one +/// message chunk, the tool updates, holds the turn open, and finally responds. +fn adapter_script() -> String { + format!( + r#"#!/usr/bin/env node +import readline from "node:readline"; + +const lines = readline.createInterface({{ input: process.stdin }}); +const send = (message) => console.log(JSON.stringify(message)); +const update = (update) => + send({{ + jsonrpc: "2.0", + method: "session/update", + params: {{ sessionId: "adapter-session", update }}, + }}); + +for await (const line of lines) {{ + if (!line.trim()) continue; + const message = JSON.parse(line); + if (message.method === "initialize") {{ + send({{ + jsonrpc: "2.0", + id: message.id, + result: {{ + protocolVersion: message.params.protocolVersion, + agentInfo: {{ name: "live-updates-acp-adapter" }}, + configOptions: [] + }} + }}); + }} else if (message.method === "session/new") {{ + send({{ + jsonrpc: "2.0", + id: message.id, + result: {{ + sessionId: "adapter-session", + modes: {{ currentModeId: "default", availableModes: [] }}, + models: {{ + currentModelId: "fast-model", + availableModels: [{{ modelId: "fast-model", name: "Fast Model" }}] + }} + }} + }}); + }} else if (message.method === "session/prompt") {{ + // A leading assistant chunk: this is what opens the completion buffer. + update({{ + sessionUpdate: "agent_message_chunk", + content: {{ type: "text", text: "banner\n" }} + }}); + update({{ + sessionUpdate: "tool_call", + toolCallId: "tool-1", + title: "sleep", + kind: "execute", + status: "pending", + rawInput: {{ command: "sleep 60" }} + }}); + update({{ + sessionUpdate: "tool_call_update", + toolCallId: "tool-1", + title: "sleep", + kind: "execute", + status: "in_progress" + }}); + // The tool is "running": nothing else is produced until it finishes. + await new Promise((resolve) => setTimeout(resolve, {hold})); + update({{ + sessionUpdate: "tool_call_update", + toolCallId: "tool-1", + status: "completed" + }}); + send({{ jsonrpc: "2.0", id: message.id, result: {{ stopReason: "end_turn" }} }}); + }} else {{ + send({{ + jsonrpc: "2.0", + id: message.id, + error: {{ code: -32601, message: `unknown method ${{message.method}}` }} + }}); + }} +}} +"#, + hold = ADAPTER_HOLD_MS, + ) +} + +fn dispatch_acp( + sidecar: &mut NativeSidecar, + request_id: i64, + connection_id: &str, + session_id: &str, + vm_id: &str, + request: AcpRequest, +) -> AcpResponse { + let payload = serde_bare::to_vec(&request).expect("encode ACP request"); + let result = sidecar + .dispatch_wire_blocking(RequestFrame { + schema: agentos_native_sidecar::wire::protocol_schema(), + request_id, + ownership: OwnershipScope::VmOwnership(VmOwnership { + connection_id: connection_id.to_owned(), + session_id: session_id.to_owned(), + vm_id: vm_id.to_owned(), + }), + payload: RequestPayload::ExtEnvelope(ExtEnvelope { + namespace: String::from(ACP_EXTENSION_NAMESPACE), + payload, + }), + }) + .expect("dispatch ACP extension request"); + + match result.response.payload { + ResponsePayload::ExtEnvelope(envelope) => { + assert_eq!(envelope.namespace, ACP_EXTENSION_NAMESPACE); + serde_bare::from_slice(&envelope.payload).expect("decode ACP response") + } + ResponsePayload::RejectedResponse(rejected) => panic!( + "ACP dispatch was rejected at the wire layer: code={} message={}", + rejected.code, rejected.message + ), + other => panic!("unexpected sidecar response: {other:?}"), + } +} + +fn assert_node_available() { + let output = Command::new("node") + .arg("--version") + .output() + .expect("spawn node --version"); + assert!(output.status.success(), "node must be available"); +} + +fn new_sidecar(name: &str) -> NativeSidecar { + NativeSidecar::with_config_and_extensions( + RecordingBridge::default(), + NativeSidecarConfig { + sidecar_id: format!("sidecar-{name}"), + compile_cache_root: Some(temp_dir(name).join("cache")), + ..NativeSidecarConfig::default() + }, + agentos_sidecar_wrapper::extensions(), + ) + .expect("create native sidecar") +} + +fn authenticate(sidecar: &mut NativeSidecar) -> String { + let result = sidecar + .dispatch_wire_blocking(RequestFrame { + schema: agentos_native_sidecar::wire::protocol_schema(), + request_id: 1, + ownership: OwnershipScope::ConnectionOwnership(ConnectionOwnership { + connection_id: String::from("client"), + }), + payload: RequestPayload::AuthenticateRequest(AuthenticateRequest { + client_name: String::from("acp-extension-live-updates"), + auth_token: String::new(), + protocol_version: agentos_native_sidecar::wire::PROTOCOL_VERSION, + bridge_version: agentos_bridge::bridge_contract().version, + }), + }) + .expect("authenticate"); + match result.response.payload { + ResponsePayload::AuthenticatedResponse(response) => response.connection_id, + other => panic!("unexpected auth response: {other:?}"), + } +} + +fn open_session(sidecar: &mut NativeSidecar, connection_id: &str) -> String { + let result = sidecar + .dispatch_wire_blocking(RequestFrame { + schema: agentos_native_sidecar::wire::protocol_schema(), + request_id: 2, + ownership: OwnershipScope::ConnectionOwnership(ConnectionOwnership { + connection_id: connection_id.to_owned(), + }), + payload: RequestPayload::OpenSessionRequest(OpenSessionRequest { + placement: SidecarPlacement::SidecarPlacementShared(SidecarPlacementShared { + pool: None, + }), + metadata: HashMap::new(), + }), + }) + .expect("open session"); + match result.response.payload { + ResponsePayload::SessionOpenedResponse(response) => response.session_id, + other => panic!("unexpected session response: {other:?}"), + } +} + +fn create_vm( + sidecar: &mut NativeSidecar, + connection_id: &str, + session_id: &str, + cwd: &Path, +) -> String { + let result = sidecar + .dispatch_wire_blocking(RequestFrame { + schema: agentos_native_sidecar::wire::protocol_schema(), + request_id: 3, + ownership: OwnershipScope::SessionOwnership(SessionOwnership { + connection_id: connection_id.to_owned(), + session_id: session_id.to_owned(), + }), + payload: RequestPayload::CreateVmRequest(CreateVmRequest { + runtime: GuestRuntimeKind::JavaScript, + config: serde_json::to_string(&vm_config::CreateVmConfig { + cwd: Some(cwd.to_string_lossy().into_owned()), + database: Some(vm_config::VmSqliteDescriptor::SqliteFile { + path: cwd.join("agentos.sqlite").to_string_lossy().into_owned(), + }), + permissions: Some(allow_all_permissions()), + ..Default::default() + }) + .expect("serialize create VM config"), + }), + }) + .expect("create VM"); + let vm_id = match result.response.payload { + ResponsePayload::VmCreatedResponse(response) => response.vm_id, + other => panic!("unexpected create VM response: {other:?}"), + }; + configure_mock_agent_package(sidecar, connection_id, session_id, &vm_id, cwd); + vm_id +} + +fn configure_mock_agent_package( + sidecar: &mut NativeSidecar, + connection_id: &str, + session_id: &str, + vm_id: &str, + cwd: &Path, +) { + let script = fs::read_to_string(cwd.join(ADAPTER_FILE)).expect("read adapter script"); + let package_dir = cwd.join("packages").join("pi"); + let bin_dir = package_dir.join("bin"); + fs::create_dir_all(&bin_dir).expect("create mock agent bin dir"); + let manifest = serde_json::json!({ + "name": "pi", + "version": "0.0.0", + "agent": { "acpEntrypoint": "pi" }, + }) + .to_string(); + fs::write(package_dir.join("agentos-package.json"), manifest) + .expect("write mock agent manifest"); + let command = bin_dir.join("pi"); + fs::write(&command, script).expect("write mock agent command"); + fs::set_permissions(&command, fs::Permissions::from_mode(0o755)) + .expect("make mock agent command executable"); + let result = sidecar + .dispatch_wire_blocking(RequestFrame { + schema: agentos_native_sidecar::wire::protocol_schema(), + request_id: 30, + ownership: OwnershipScope::VmOwnership(VmOwnership { + connection_id: connection_id.to_owned(), + session_id: session_id.to_owned(), + vm_id: vm_id.to_owned(), + }), + payload: RequestPayload::ConfigureVmRequest(ConfigureVmRequest { + mounts: Vec::new(), + software: Vec::new(), + permissions: None, + module_access_cwd: None, + instructions: Vec::new(), + projected_modules: Vec::new(), + command_permissions: HashMap::new(), + loopback_exempt_ports: Vec::new(), + packages: vec![PackageDescriptor { + path: package_dir.to_string_lossy().into_owned(), + }], + packages_mount_at: String::from("/opt/agentos"), + bootstrap_commands: Vec::new(), + binding_shim_commands: Vec::new(), + }), + }) + .expect("configure mock ACP package"); + assert!(matches!( + result.response.payload, + ResponsePayload::VmConfiguredResponse(_) + )); +} + +fn allow_all_permissions() -> vm_config::PermissionsPolicy { + vm_config::PermissionsPolicy { + fs: Some(vm_config::FsPermissionScope::Mode( + vm_config::PermissionMode::Allow, + )), + network: Some(vm_config::PatternPermissionScope::Mode( + vm_config::PermissionMode::Allow, + )), + child_process: Some(vm_config::PatternPermissionScope::Mode( + vm_config::PermissionMode::Allow, + )), + process: Some(vm_config::PatternPermissionScope::Mode( + vm_config::PermissionMode::Allow, + )), + env: Some(vm_config::PatternPermissionScope::Mode( + vm_config::PermissionMode::Allow, + )), + binding: Some(vm_config::PatternPermissionScope::Mode( + vm_config::PermissionMode::Allow, + )), + } +} + +fn temp_dir(name: &str) -> PathBuf { + let root = std::env::temp_dir().join(format!( + "agentos-sidecar-{name}-{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time before unix epoch") + .as_nanos() + )); + fs::create_dir_all(&root).expect("create temp dir"); + root +} diff --git a/crates/native-sidecar/src/state.rs b/crates/native-sidecar/src/state.rs index 8675b372be..3dfca8c2e5 100644 --- a/crates/native-sidecar/src/state.rs +++ b/crates/native-sidecar/src/state.rs @@ -1770,20 +1770,26 @@ impl SocketReadinessSubscribers { .any(|subscriber| subscriber.application_read_interest)) } - fn unregister(&self, identity: (u64, u64)) -> bool { + /// Drop one alias' subscription, returning the remaining aggregate read + /// interest and the removed target so the caller can also retire the + /// capability in that VM's readiness broker. + fn unregister(&self, identity: (u64, u64)) -> (bool, Option) { self.subscribers .lock() .map(|mut subscribers| { - subscribers.remove(&identity); - subscribers - .values() - .any(|subscriber| subscriber.application_read_interest) + let removed = subscribers.remove(&identity); + ( + subscribers + .values() + .any(|subscriber| subscriber.application_read_interest), + removed.map(|subscriber| subscriber.target), + ) }) .unwrap_or_else(|_| { eprintln!( "ERR_AGENTOS_READY_STATE_POISONED: socket readiness subscriber lock poisoned" ); - false + (false, None) }) } @@ -1967,8 +1973,23 @@ impl Drop for SocketReadinessRegistration { .unwrap_or_else(|error| error.into_inner()) .take(); if let Some(identity) = identity { - let aggregate = self.subscribers.unregister(identity); + let (aggregate, removed) = self.subscribers.unregister(identity); self.update_aggregate_interest(aggregate); + // The guest calls `unregisterCapabilityReadiness` on the same + // teardown, so its dispatch target is gone. Retire the capability + // in the broker too: level flags left behind have no reachable + // guest target and keep the session's wake lane armed. + if let Some(target) = removed { + if let Err(error) = target + .session + .remove_readiness(target.capability_id, target.capability_generation) + { + eprintln!( + "ERR_AGENTOS_NET_SOCKET_READY_REMOVE: capability={} generation={} teardown: {error}", + target.capability_id, target.capability_generation + ); + } + } } } } @@ -3043,4 +3064,39 @@ mod socket_readiness_registry_tests { drop(parent); assert!(subscribers.targets().is_empty()); } + + #[test] + fn alias_teardown_surfaces_its_capability_for_broker_removal() { + let process_runtime = + agentos_runtime::SidecarRuntime::process(&agentos_runtime::RuntimeConfig::default()) + .expect("create teardown test runtime"); + let resources = ResourceLedger::root( + "socket-teardown-test", + [( + ResourceClass::Capabilities, + ResourceLimit::new(2, "limits.reactor.maxCapabilities"), + )], + ); + let host = + V8RuntimeHost::spawn(&process_runtime.context()).expect("spawn teardown test V8 host"); + let session = host.session_handle(String::from("socket-teardown-test")); + let subscribers = SocketReadinessSubscribers::new(&resources); + subscribers + .register(None, javascript_target(&session, 7)) + .expect("register alias"); + + let (aggregate, removed) = subscribers.unregister((7, 1)); + assert!(!aggregate); + let removed = removed.expect( + "teardown must surface the target so its capability can be retired in the broker", + ); + assert_eq!( + (removed.capability_id, removed.capability_generation), + (7, 1) + ); + assert!( + subscribers.unregister((7, 1)).1.is_none(), + "a repeated teardown must not retire the capability twice" + ); + } } diff --git a/crates/v8-runtime/src/isolate.rs b/crates/v8-runtime/src/isolate.rs index 11acea69d7..508d1f498d 100644 --- a/crates/v8-runtime/src/isolate.rs +++ b/crates/v8-runtime/src/isolate.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use std::ffi::c_void; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{mpsc, Mutex, Once}; use std::thread; @@ -169,9 +170,9 @@ pub fn prepare_current_thread() { // Headroom granted to V8 when the near-heap-limit callback fires. V8 fatal-aborts // the whole process (SIGTRAP) if the callback does not raise the limit, so we must -// hand back a larger limit to give the engine room to unwind. Termination has -// already been requested, so this extra budget only covers propagation of the -// uncatchable termination exception, not continued guest allocation. +// hand back a larger limit to give the engine room to unwind, and it also gives a +// guest that merely peaked at its cap one chance to fall back under it before the +// isolate is terminated. const NEAR_HEAP_LIMIT_HEADROOM_BYTES: usize = 16 * 1024 * 1024; /// Default per-isolate heap cap applied when the caller passes no explicit limit. @@ -183,35 +184,65 @@ const NEAR_HEAP_LIMIT_HEADROOM_BYTES: usize = 16 * 1024 * 1024; /// isolation semantics; operators may raise it via the configured limit. pub const DEFAULT_HEAP_LIMIT_MB: u32 = 128; +/// Per-isolate state behind the near-heap-limit callback's `data` pointer. +struct HeapLimitGuard { + /// Thread-safe handle used to terminate this very isolate. + handle: v8::IsolateHandle, + /// Set once the callback has already raised this isolate's limit by + /// [`NEAR_HEAP_LIMIT_HEADROOM_BYTES`]. + headroom_granted: AtomicBool, +} + +/// True when the guest must be terminated: the headroom was already granted for +/// this isolate and the heap reached the raised limit anyway. +fn heap_guard_should_terminate(headroom_granted: &AtomicBool) -> bool { + headroom_granted.swap(true, Ordering::SeqCst) +} + /// Invoked by V8 when heap usage approaches the configured limit. Instead of -/// letting V8 fatal-abort the (process-global) runtime, request termination of the -/// offending isolate and return a raised limit so V8 can propagate the uncatchable -/// termination exception cleanly. `data` is a leaked `Box` for -/// the isolate this callback was registered on. +/// letting V8 fatal-abort the (process-global) runtime, raise the limit and — for +/// a guest that cannot fit inside that headroom — terminate the offending isolate. +/// `data` is a leaked `Box` for the isolate this callback was +/// registered on. +/// +/// The first callback is NOT a termination. V8 raises it as soon as a full GC +/// cannot fit the live set under the cap, which a healthy guest sitting at its +/// high-water mark reaches without being a heap bomb. Terminating there kills +/// guest JS mid-frame: the termination exception is uncatchable, skips `finally`, +/// and — outside an `Execute` that would cancel it — leaves the isolate in the +/// terminating state, where every later guest callback silently returns nothing. +/// A session in that state does not fail; it goes idle forever. So grant the +/// headroom first and terminate only if the heap reaches the raised limit too, +/// which keeps the guest bounded at `limit + NEAR_HEAP_LIMIT_HEADROOM_BYTES`. extern "C" fn near_heap_limit_callback( data: *mut c_void, current_heap_limit: usize, initial_heap_limit: usize, ) -> usize { - if !data.is_null() { - // Safety: `data` is the pointer produced by `Box::into_raw` in - // `install_heap_limit_guard` and lives for the entire lifetime of the - // isolate. - let handle = unsafe { &*(data as *const v8::IsolateHandle) }; - // Terminate any JS currently running on this isolate. This unwinds the - // guest with an uncatchable exception rather than crashing the process. - handle.terminate_execution(); + // Never shrink below the current limit: V8 fatal-aborts if this callback + // hands back a limit the heap is already over. + let raised = current_heap_limit + .max(initial_heap_limit) + .saturating_add(NEAR_HEAP_LIMIT_HEADROOM_BYTES); + if data.is_null() { + return raised; + } + // Safety: `data` is the pointer produced by `Box::into_raw` in + // `install_heap_limit_guard` and lives for the entire lifetime of the + // isolate. + let guard = unsafe { &*(data as *const HeapLimitGuard) }; + if !heap_guard_should_terminate(&guard.headroom_granted) { + return raised; } + // Terminate any JS currently running on this isolate. This unwinds the + // guest with an uncatchable exception rather than crashing the process. + guard.handle.terminate_execution(); warn_limit_exhausted( TrackedLimit::V8HeapBytes, current_heap_limit, initial_heap_limit.max(1), ); - // Grant headroom so V8 does not immediately fatal-abort before the termination - // takes effect. We never shrink below the current limit. - current_heap_limit - .max(initial_heap_limit) - .saturating_add(NEAR_HEAP_LIMIT_HEADROOM_BYTES) + raised } /// Register the near-heap-limit OOM guard on an isolate that was created with a @@ -223,11 +254,14 @@ extern "C" fn near_heap_limit_callback( /// regardless of whether it was built fresh or restored from a snapshot. pub fn install_heap_limit_guard(isolate: &mut v8::OwnedIsolate) { // The callback needs a thread-safe handle to request termination of this very - // isolate. The handle is leaked so it outlives the callback registration; the + // isolate. The guard is leaked so it outlives the callback registration; the // number of isolates per process is bounded, so this is not an unbounded leak, // and the memory is reclaimed when the process exits. - let handle = Box::new(isolate.thread_safe_handle()); - let data = Box::into_raw(handle) as *mut c_void; + let guard = Box::new(HeapLimitGuard { + handle: isolate.thread_safe_handle(), + headroom_granted: AtomicBool::new(false), + }); + let data = Box::into_raw(guard) as *mut c_void; isolate.add_near_heap_limit_callback(near_heap_limit_callback, data); } @@ -276,3 +310,21 @@ pub fn create_context(isolate: &mut v8::OwnedIsolate) -> v8::Global // V8 lifecycle tests are consolidated in execution::tests to avoid // inter-test SIGSEGV from V8 global state issues. + +#[cfg(test)] +mod tests { + use super::{heap_guard_should_terminate, AtomicBool}; + + /// The first near-heap-limit callback must only raise the limit. Terminating + /// there kills a guest that merely peaked at its cap, and a termination that + /// lands outside an `Execute` is never cancelled: the isolate then answers + /// every guest callback with nothing and the session idles forever instead of + /// failing. + #[test] + fn heap_guard_grants_headroom_once_before_terminating() { + let headroom_granted = AtomicBool::new(false); + assert!(!heap_guard_should_terminate(&headroom_granted)); + assert!(heap_guard_should_terminate(&headroom_granted)); + assert!(heap_guard_should_terminate(&headroom_granted)); + } +} diff --git a/crates/v8-runtime/src/session.rs b/crates/v8-runtime/src/session.rs index c5139ba709..229444050c 100644 --- a/crates/v8-runtime/src/session.rs +++ b/crates/v8-runtime/src/session.rs @@ -67,6 +67,11 @@ struct SessionReadiness { broker: RuntimeSessionReadyBroker, wakes: Mutex, executor_wake_tx: Sender, + /// Capability identities whose last guest dispatch found no registered + /// target, keyed to the revision that missed. Entries are dropped on + /// delivery, on acknowledgement and on capability removal, so this holds at + /// most the capabilities currently between two readiness wakes. + target_misses: Mutex>, } impl SessionReadiness { @@ -94,6 +99,7 @@ impl SessionReadiness { broker, wakes: Mutex::new(SessionReadyWakeState { runtime_wake_rx }), executor_wake_tx, + target_misses: Mutex::new(HashMap::new()), }), executor_wake_rx, )) @@ -113,6 +119,7 @@ impl SessionReadiness { broker, wakes: Mutex::new(SessionReadyWakeState { runtime_wake_rx }), executor_wake_tx, + target_misses: Mutex::new(HashMap::new()), }), executor_wake_rx, )) @@ -144,6 +151,7 @@ impl SessionReadiness { } fn remove(&self, capability_id: u64, capability_generation: u64) -> Result<(), String> { + self.forget_target_misses((capability_id, capability_generation)); self.broker .remove_capability(self.generation, capability_id, capability_generation) .map_err(|error| error.to_string()) @@ -214,6 +222,51 @@ impl SessionReadiness { .map_err(|error| error.to_string()) } + /// Record one guest dispatch outcome, appending the observation to + /// `delivered` when the broker must clear its level flags for it. + /// + /// A missing guest target gets exactly one free retry, which covers the + /// genuine race this tolerates: readiness published before the guest + /// finished registering its capability resolves within a single wake. The + /// broker is level-triggered, so an observation that is never acknowledged + /// keeps `complete_wake` rearming immediately, and a target that will never + /// appear — teardown already unregistered it, or its generation is + /// permanently stale — spins the session at full CPU without ever reaching + /// a turn boundary. Acknowledging the second consecutive miss of the same + /// revision clears the flags and lets the loop idle. A capability that + /// registers afterwards still receives delivery: new data bumps the + /// revision and republishes readiness. + fn record_dispatch_outcome( + &self, + entry: &ReadyObservation, + delivered_to_guest: bool, + delivered: &mut Vec, + ) { + let identity = (entry.capability_id, entry.capability_generation); + // Advisory bookkeeping: a poisoned lock must not fail the readiness + // turn, and recovering the map only costs one extra retry. + let mut misses = self + .target_misses + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()); + if delivered_to_guest { + misses.remove(&identity); + delivered.push(*entry); + return; + } + if misses.insert(identity, entry.revision) == Some(entry.revision) { + misses.remove(&identity); + delivered.push(*entry); + } + } + + fn forget_target_misses(&self, identity: (u64, u64)) { + self.target_misses + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .remove(&identity); + } + fn complete_batch( &self, batch: &RuntimeReadyBatch, @@ -3667,17 +3720,31 @@ fn dispatch_ready_batch_callbacks( entry.flags, ); tc.perform_microtask_checkpoint(); + // An out-of-band termination (the near-heap-limit guard) makes every + // guest call return nothing, which is indistinguishable from a missing + // target here. Reporting it as one would retire the capability's level + // flags against an isolate that can no longer run any JavaScript, and + // the session would idle forever instead of failing. Surface it as the + // termination it is; the caller cancels it and reports the execution. + if tc.has_terminated() { + return EventLoopStatus::Terminated; + } if let Some(exception) = tc.exception() { let (code, error) = execution::exception_to_result(tc, exception); return EventLoopStatus::Failed(code, error); } match dispatch { - crate::stream::ReadinessDispatch::Delivered => delivered.push(*entry), + crate::stream::ReadinessDispatch::Delivered => { + ready_broker.record_dispatch_outcome(entry, true, delivered); + } crate::stream::ReadinessDispatch::TargetMissing => { // The bridge exists but this capability may not be registered - // yet (for example, readiness raced the connect response). - // Leave the observation unacknowledged so the durable - // sidecar state schedules another coalesced wake. + // yet (for example, readiness raced the connect response). The + // first miss stays unacknowledged so the durable sidecar state + // schedules another coalesced wake; the second is acknowledged + // so the level-triggered broker cannot rearm forever against a + // target that will never appear. + ready_broker.record_dispatch_outcome(entry, false, delivered); } crate::stream::ReadinessDispatch::BridgeMissing => { return EventLoopStatus::Failed( @@ -3926,6 +3993,7 @@ fn dispatch_event_loop_frame( #[cfg(test)] mod tests { use super::*; + use agentos_runtime::metrics::WakeMetric; use std::collections::HashSet; const TEST_READY_BATCH_HANDLES: usize = 64; @@ -4607,6 +4675,76 @@ mod tests { ); } + /// Regression test for the readiness livelock: a capability whose guest + /// dispatch target is missing must not rearm the level-triggered broker + /// forever. Drives the same bookkeeping `dispatch_ready_batch_callbacks` + /// uses, without a V8 isolate, and asserts the wake counters stop growing. + #[test] + fn missing_readiness_target_stops_rearming_after_one_free_retry() { + let mgr = test_manager(1); + let metrics = mgr.runtime.metrics().clone(); + let wake_count = || { + let snapshot = metrics.snapshot(); + snapshot.wakes[WakeMetric::Delivered.index()] + + snapshot.wakes[WakeMetric::Rearmed.index()] + }; + let (broker, wake_rx) = SessionReadiness::new(41, &mgr.runtime, TEST_READY_BATCH_HANDLES) + .expect("create session readiness"); + broker + .publish(77, 5, ReadyFlags::READABLE) + .expect("publish readiness with no guest target registered"); + + let mut turns = 0; + while let Ok(wake) = wake_rx.try_recv() { + let batch = broker.take_batch(wake).expect("readiness batch"); + let mut delivered = Vec::new(); + for entry in &batch.entries { + broker.record_dispatch_outcome(entry, false, &mut delivered); + } + broker + .complete_batch(&batch, &delivered) + .expect("complete readiness wake"); + turns += 1; + assert!( + turns <= 8, + "an unacknowledgeable readiness observation rearmed without bound" + ); + } + assert_eq!(turns, 2, "a missing target gets exactly one free retry"); + assert_eq!( + broker.broker.pending_handle_count().expect("pending count"), + 0, + "the bounded miss must clear the retained level flags" + ); + + let bounded = wake_count(); + assert!(wake_rx.try_recv().is_err(), "the wake lane must go quiet"); + assert_eq!(wake_count(), bounded, "wake metrics must stop growing"); + + // A capability that registers afterwards still receives delivery: new + // data bumps the revision and republishes readiness. + broker + .publish(77, 5, ReadyFlags::READABLE) + .expect("republish readiness after the guest registered its target"); + let wake = wake_rx + .try_recv() + .expect("republished readiness must wake the session"); + let batch = broker.take_batch(wake).expect("post-registration batch"); + let mut delivered = Vec::new(); + for entry in &batch.entries { + broker.record_dispatch_outcome(entry, true, &mut delivered); + } + assert_eq!(delivered, batch.entries); + broker + .complete_batch(&batch, &delivered) + .expect("acknowledge delivered readiness"); + assert!(wake_rx.try_recv().is_err()); + assert!( + wake_count() > bounded, + "delivery must resume once the guest target exists" + ); + } + #[test] fn readiness_dispatch_failure_completes_wake_before_session_reuse() { let mgr = test_manager(1);