From b461a0dff20c967162fbb8d2014dadca1d71caf7 Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Tue, 7 Jul 2026 19:23:43 -0700 Subject: [PATCH 1/4] fix(agentos-sidecar): harden the close_session already-exited short-circuit - Builds on #1640: also short-circuit on the session record already marked closed (covers exits the kill RPC does not report), and skip the second wait when SIGKILL itself reports the process gone. - Mirror the short-circuit in the agentos-sidecar-core sync engine, with regression tests proving zero dead waits. --- crates/agentos-sidecar-core/src/engine.rs | 134 +++++++++++++++++++- crates/agentos-sidecar/src/acp_extension.rs | 64 +++++----- 2 files changed, 160 insertions(+), 38 deletions(-) diff --git a/crates/agentos-sidecar-core/src/engine.rs b/crates/agentos-sidecar-core/src/engine.rs index 3d1235d40..8c7ab2558 100644 --- a/crates/agentos-sidecar-core/src/engine.rs +++ b/crates/agentos-sidecar-core/src/engine.rs @@ -227,13 +227,25 @@ impl AcpCore { }; let _ = host.close_stdin(&session.process_id); - let _ = host.kill_agent(&session.process_id, "SIGTERM"); - if host - .wait_for_exit(&session.process_id, SESSION_CLOSE_TIMEOUT_MS)? - .is_none() + // Mirror of the native `close_session` short-circuit: an adapter that + // already exited (crash / OOM / idle eviction, recorded on the session + // as `closed`) has no future exit to wait for, and signalling its + // reaped PID fails with a process-gone error. Skip the SIGTERM → wait + // → SIGKILL → wait dance (~2× `SESSION_CLOSE_TIMEOUT_MS` of dead + // waiting) in that case. + let adapter_already_gone = session.closed || { + let sigterm = host.kill_agent(&session.process_id, "SIGTERM"); + matches!(&sigterm, Err(error) if is_process_already_gone_error(error)) + }; + if !adapter_already_gone + && host + .wait_for_exit(&session.process_id, SESSION_CLOSE_TIMEOUT_MS)? + .is_none() { - let _ = host.kill_agent(&session.process_id, "SIGKILL"); - let _ = host.wait_for_exit(&session.process_id, SESSION_CLOSE_TIMEOUT_MS)?; + let sigkill = host.kill_agent(&session.process_id, "SIGKILL"); + if !matches!(&sigkill, Err(error) if is_process_already_gone_error(error)) { + let _ = host.wait_for_exit(&session.process_id, SESSION_CLOSE_TIMEOUT_MS)?; + } } Ok(AcpResponse::AcpSessionClosedResponse( @@ -1121,6 +1133,18 @@ fn is_unknown_session_error(response: &Value) -> bool { .is_some_and(|kind| kind == "unknown_session") } +/// True when a signal/kill request failed because the target process no longer +/// exists — the process-table `ESRCH` / "no such process" / "has no active +/// process" errors surfaced for an already-reaped PID. `close_session` uses +/// this to skip the exit wait when the adapter is already gone. Mirrors the +/// native sidecar's `is_process_already_gone_error`. +fn is_process_already_gone_error(error: &AcpCoreError) -> bool { + let message = error.to_string().to_ascii_lowercase(); + message.contains("esrch") + || message.contains("no such process") + || message.contains("has no active process") +} + /// Write a JSON-RPC message as a single newline-terminated line to the agent's /// stdin (no waiting). Used by the resumable handshake. fn write_json_line( @@ -1329,6 +1353,104 @@ mod tests { assert_eq!(host.killed, vec![("proc-s1".into(), "SIGTERM".into())]); } + /// Host whose adapter process is already gone: `wait_for_exit` would time + /// out (returns `None`), so any wait the engine performs is dead time. The + /// wait counter proves the short-circuit: a regression re-enters the + /// SIGTERM → wait → SIGKILL → wait sequence and records 2 waits. + #[derive(Default)] + struct GoneAdapterHost { + kill_error: Option, + killed: Vec<(String, String)>, + waits: usize, + } + + impl AcpHost for GoneAdapterHost { + fn spawn_agent(&mut self, _: SpawnAgentRequest) -> Result { + unreachable!("close_session does not spawn") + } + fn bind_session(&mut self, _: &str, _: &str) -> Result<(), AcpCoreError> { + Ok(()) + } + fn write_stdin(&mut self, _: &str, _: &[u8]) -> Result<(), AcpCoreError> { + unreachable!() + } + fn close_stdin(&mut self, _: &str) -> Result<(), AcpCoreError> { + Ok(()) + } + fn poll_output(&mut self, _: &str) -> Result, AcpCoreError> { + Ok(None) + } + fn kill_agent(&mut self, process_id: &str, signal: &str) -> Result<(), AcpCoreError> { + self.killed + .push((process_id.to_string(), signal.to_string())); + match &self.kill_error { + Some(message) => Err(AcpCoreError::InvalidState(message.clone())), + None => Ok(()), + } + } + fn wait_for_exit(&mut self, _: &str, _: u64) -> Result, AcpCoreError> { + self.waits += 1; + Ok(None) // the exit event was already drained; a wait can only time out + } + fn write_file(&mut self, _: &str, _: &[u8]) -> Result<(), AcpCoreError> { + Ok(()) + } + fn read_file(&mut self, _: &str) -> Result, AcpCoreError> { + Ok(Vec::new()) + } + fn now_ms(&self) -> u64 { + 0 + } + } + + #[test] + fn close_session_skips_teardown_wait_when_session_already_closed() { + let mut core = AcpCore::new(); + let mut dead = record("s1", "conn-a"); + dead.closed = true; + dead.exit_code = Some(137); + core.insert_session(dead); + let mut host = GoneAdapterHost::default(); + let resp = core + .close_session( + &mut host, + "conn-a", + &AcpCloseSessionRequest { + session_id: "s1".into(), + }, + ) + .expect("close"); + assert!(matches!(resp, AcpResponse::AcpSessionClosedResponse(_))); + assert_eq!(core.session_count(), 0); + // Already-observed exit: no signals sent, no dead waiting. + assert!(host.killed.is_empty()); + assert_eq!(host.waits, 0); + } + + #[test] + fn close_session_skips_teardown_wait_when_sigterm_reports_process_gone() { + let mut core = AcpCore::new(); + core.insert_session(record("s1", "conn-a")); + let mut host = GoneAdapterHost { + kill_error: Some(String::from("process proc-s1: no such process (ESRCH)")), + ..GoneAdapterHost::default() + }; + let resp = core + .close_session( + &mut host, + "conn-a", + &AcpCloseSessionRequest { + session_id: "s1".into(), + }, + ) + .expect("close"); + assert!(matches!(resp, AcpResponse::AcpSessionClosedResponse(_))); + // SIGTERM was attempted, classified as process-gone, and both waits + // (and the SIGKILL escalation) were skipped. + assert_eq!(host.killed, vec![("proc-s1".into(), "SIGTERM".into())]); + assert_eq!(host.waits, 0); + } + #[test] fn session_request_enforces_ownership_without_side_effects() { // A non-owner prompt fails closed with the same unknown-session error and diff --git a/crates/agentos-sidecar/src/acp_extension.rs b/crates/agentos-sidecar/src/acp_extension.rs index 9cedadb7e..4f2ef4149 100644 --- a/crates/agentos-sidecar/src/acp_extension.rs +++ b/crates/agentos-sidecar/src/acp_extension.rs @@ -626,37 +626,38 @@ impl AcpExtension { process_id: session.process_id.clone(), }) .await; - let sigterm = ctx - .kill_process_wire(KillProcessRequest { - process_id: session.process_id.clone(), - signal: String::from("SIGTERM"), - }) - .await; - // The adapter process may already be gone: it can exit out-of-band (crash / - // OOM, or a lazily-observed idle-time crash) *before* the client sends - // close_session. When it does, its `ProcessExited` event has already been - // emitted and drained, so `wait_for_process_exit` — which only polls the - // event stream and never checks current process state — can never observe - // it and blocks the full `SESSION_CLOSE_TIMEOUT` for SIGTERM *and* again - // after SIGKILL (~2× the timeout) before we ever reach resource disposal. - // Because close_session is serial per sidecar, that stall also blocks a - // `create_session` the host issues right after it (session recovery / a - // returning user whose idle session was evicted eats the full stall on - // their next turn). Detect the already-gone case from the SIGTERM result — - // reusing the same "adapter is gone" classification the prompt path uses — - // and skip straight to resource disposal. - let already_gone = matches!(&sigterm, Err(error) if is_process_already_gone(error)); - if !already_gone + // The adapter may already be gone: it can crash, OOM, or idle-evict + // before the client sends close_session, and its `ProcessExitedEvent` + // has then already been drained from the shared per-ownership event + // queue (usually by the prompt exchange loop, which records it as + // `session.closed`). `wait_for_process_exit` only observes *future* + // events, so without a short-circuit an already-dead adapter burns + // `SESSION_CLOSE_TIMEOUT` twice (~10s) signalling a PID that no longer + // exists — and because extension dispatch is serialized, a + // `create_session` issued right after (session recovery for a + // returning user) stalls behind the dead wait. + let adapter_already_gone = session.closed || { + let sigterm = ctx + .kill_process_wire(KillProcessRequest { + process_id: session.process_id.clone(), + signal: String::from("SIGTERM"), + }) + .await; + matches!(&sigterm, Err(error) if is_process_already_gone_error(error)) + }; + if !adapter_already_gone && !wait_for_process_exit(&mut ctx, &session.process_id, SESSION_CLOSE_TIMEOUT).await { - let _ = ctx + let sigkill = ctx .kill_process_wire(KillProcessRequest { process_id: session.process_id.clone(), signal: String::from("SIGKILL"), }) .await; - let _ = - wait_for_process_exit(&mut ctx, &session.process_id, SESSION_CLOSE_TIMEOUT).await; + if !matches!(&sigkill, Err(error) if is_process_already_gone_error(error)) { + let _ = wait_for_process_exit(&mut ctx, &session.process_id, SESSION_CLOSE_TIMEOUT) + .await; + } } let _ = ctx .dispose_session_resources_wire(&request.session_id) @@ -2571,14 +2572,13 @@ fn is_adapter_gone_error(error: &SidecarError) -> bool { matches!(error, SidecarError::InvalidState(message) if message.contains(ADAPTER_NO_ACTIVE_PROCESS_MARKER)) } -/// True when a kill/signal request failed because the target process no longer -/// exists — i.e. it already exited before we tried to terminate it. Covers both -/// the adapter-gone classification (`is_adapter_gone_error`) and the lower-level -/// process-table `ESRCH` / "no such process" the signal path returns for a PID -/// that has already been reaped. Used by `close_session` to skip the -/// `wait_for_process_exit` dance (which can only observe a *future* exit event) -/// when the process is already gone. -fn is_process_already_gone(error: &SidecarError) -> bool { +/// True when a signal/kill request failed because the target process no longer +/// exists: either the adapter-gone classification the prompt path uses +/// (`is_adapter_gone_error`) or the lower-level process-table `ESRCH` / +/// "no such process" error the signal path returns for an already-reaped PID. +/// `close_session` uses this to skip `wait_for_process_exit` — which can only +/// observe a *future* exit event — when the process is already gone. +fn is_process_already_gone_error(error: &SidecarError) -> bool { if is_adapter_gone_error(error) { return true; } From 190fdca5e9e3254bc83feec14731ef378646bbca Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Tue, 7 Jul 2026 19:27:11 -0700 Subject: [PATCH 2/4] fix(native-sidecar): resolve js_bridge mount-root realpath without a bridge round-trip - Resolve `realpath("/")` of a js_bridge mount locally (mirroring host_dir): host-side drivers often cannot canonicalize their own root and returned ENOENT, which aborted kernel permission-subject resolution before the readDir bridge call was ever issued. - Regression test: readdir of a js_bridge mount root against a driver whose realpath always fails. --- .../native-sidecar/src/plugins/js_bridge.rs | 12 ++ crates/native-sidecar/tests/service.rs | 114 +++++++++++++++++- 2 files changed, 121 insertions(+), 5 deletions(-) diff --git a/crates/native-sidecar/src/plugins/js_bridge.rs b/crates/native-sidecar/src/plugins/js_bridge.rs index 06c98f6bb..1391b74b3 100644 --- a/crates/native-sidecar/src/plugins/js_bridge.rs +++ b/crates/native-sidecar/src/plugins/js_bridge.rs @@ -517,6 +517,18 @@ impl VirtualFileSystem for JsBridgeFilesystem { } fn realpath(&self, path: &str) -> VfsResult { + // The mount root always exists — it IS the mount point — so resolve it + // locally like `host_dir` does instead of round-tripping the bridge. + // Host-side drivers frequently cannot canonicalize their own root and + // return ENOENT for it, and because a js_bridge mount root is not a + // symlink leaf, `MountTable::realpath` propagates that ENOENT instead + // of falling back to the lexical path. The kernel permission wrapper + // resolves every operation's subject through `realpath` first, so that + // ENOENT aborts e.g. a readdir of the mount root before the readDir + // bridge call is ever issued. + if path.is_empty() || path == "/" { + return Ok(String::from("/")); + } self.parse_required( "realpath", path, diff --git a/crates/native-sidecar/tests/service.rs b/crates/native-sidecar/tests/service.rs index 6ea4daf03..cf0c6fbfb 100644 --- a/crates/native-sidecar/tests/service.rs +++ b/crates/native-sidecar/tests/service.rs @@ -992,6 +992,19 @@ ykAheWCsAteSEWVc0w==\n\ ) -> ( Arc>, Arc>>, + ) { + install_memory_js_bridge_handler_with_options(sidecar, false) + } + + /// `fail_realpath` simulates host-side drivers that cannot canonicalize + /// their own paths and answer every `realpath` bridge call with ENOENT + /// — the shape that used to break readdir of a js_bridge mount root. + fn install_memory_js_bridge_handler_with_options( + sidecar: &mut NativeSidecar, + fail_realpath: bool, + ) -> ( + Arc>, + Arc>>, ) { let filesystem = Arc::new(Mutex::new(MemoryFileSystem::new())); let calls = Arc::new(Mutex::new(Vec::::new())); @@ -1111,11 +1124,15 @@ ykAheWCsAteSEWVc0w==\n\ .map_err(|error| format!("{}: {error}", error.code())) } "realpath" => { - let path = call_args["path"].as_str().expect("realpath path"); - filesystem - .realpath(path) - .map(|resolved| Some(json!(resolved))) - .map_err(|error| format!("{}: {error}", error.code())) + if fail_realpath { + Err(String::from("ENOENT: no such file or directory")) + } else { + let path = call_args["path"].as_str().expect("realpath path"); + filesystem + .realpath(path) + .map(|resolved| Some(json!(resolved))) + .map_err(|error| format!("{}: {error}", error.code())) + } } "symlink" => { let target = call_args["target"].as_str().expect("symlink target"); @@ -7458,6 +7475,92 @@ ykAheWCsAteSEWVc0w==\n\ .expect_err("stat should fail"); assert_eq!(stat_error.code(), "EIO"); } + + fn configure_vm_js_bridge_mount_readdir_of_mount_root_survives_broken_driver_realpath() { + // Regression: readdir of a js_bridge mount root used to fail with + // ENOENT before any readDir bridge call was issued. The kernel + // permission wrapper resolves every subject through `realpath` + // first, and host-side drivers that cannot canonicalize their own + // root answer the mount-root realpath with ENOENT — which + // `MountTable::realpath` propagates for non-symlink-leaf mounts. + // The mount root must resolve locally, without a bridge realpath. + let mut sidecar = create_test_sidecar(); + let (filesystem, calls) = + install_memory_js_bridge_handler_with_options(&mut sidecar, true); + filesystem + .lock() + .expect("lock js bridge fs") + .write_file("/hello.txt", b"hi".to_vec()) + .expect("seed js bridge fs"); + + let (connection_id, session_id) = + authenticate_and_open_session(&mut sidecar).expect("authenticate and open session"); + let vm_id = create_vm( + &mut sidecar, + &connection_id, + &session_id, + PermissionsPolicy::allow_all(), + ) + .expect("create vm"); + + sidecar + .dispatch_blocking(request( + 4, + OwnershipScope::vm(&connection_id, &session_id, &vm_id), + RequestPayload::ConfigureVm(ConfigureVmRequest { + mounts: vec![MountDescriptor { + guest_path: String::from("/workspace"), + read_only: false, + plugin: MountPluginDescriptor { + id: String::from("js_bridge"), + config: json!({ "mountId": "mount-root" }).to_string(), + }, + }], + software: Vec::new(), + permissions: None, + module_access_cwd: None, + instructions: Vec::new(), + projected_modules: Vec::new(), + command_permissions: std::collections::HashMap::new(), + loopback_exempt_ports: Vec::new(), + packages: Vec::new(), + packages_mount_at: String::new(), + bootstrap_commands: Vec::new(), + tool_shim_commands: Vec::new(), + }), + )) + .expect("configure js_bridge mount"); + + let vm = sidecar.vms.get_mut(&vm_id).expect("configured vm"); + let entries = vm + .kernel + .filesystem_mut() + .read_dir("/workspace") + .expect("readdir of js_bridge mount root"); + assert!( + entries.iter().any(|entry| entry == "hello.txt"), + "mount-root readdir should list seeded entries, got {entries:?}" + ); + + let calls = calls.lock().expect("lock js bridge calls"); + assert!( + calls + .iter() + .any(|call| call.operation == "readDir" + || call.operation == "readDirWithTypes"), + "readdir must reach the bridge; recorded operations: {:?}", + calls + .iter() + .map(|call| call.operation.clone()) + .collect::>() + ); + } + + #[test] + fn configure_vm_js_bridge_mount_root_readdir_without_bridge_realpath() { + configure_vm_js_bridge_mount_readdir_of_mount_root_survives_broken_driver_realpath(); + } + fn configure_vm_instantiates_sandbox_agent_mounts_through_the_plugin_registry() { let server = MockSandboxAgentServer::start("agentos-native-sidecar-sandbox", None); fs::write(server.root().join("hello.txt"), "hello from sandbox") @@ -21012,6 +21115,7 @@ console.log(JSON.stringify({ configure_vm_js_bridge_mount_rejects_oversized_read_payloads(); configure_vm_js_bridge_mount_rejects_pread_payloads_above_requested_length(); configure_vm_js_bridge_mount_maps_callback_errors_to_errno_codes(); + configure_vm_js_bridge_mount_readdir_of_mount_root_survives_broken_driver_realpath(); configure_vm_instantiates_sandbox_agent_mounts_through_the_plugin_registry(); configure_vm_instantiates_s3_mounts_through_the_plugin_registry(); configure_vm_instantiates_object_s3_mounts_through_the_plugin_registry(); From a46dab59596ee6213b366c81654c18d620094409 Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Tue, 7 Jul 2026 19:40:50 -0700 Subject: [PATCH 3/4] fix(core): deliver post-boot mountFs reconfiguration reliably to the native sidecar - `mountFs`/`unmountFs` now return a promise that settles once the sidecar has applied the mount; delivery failures reject instead of being swallowed. - Runtime mount reconfigures resend the boot `packages`/`packagesMountAt`/`toolShimCommands`: Rust `configure_vm` rebuilds the whole VM configuration from each payload, so omitting them stripped the `/opt/agentos` projections and tool shims as a side effect of a post-boot mount. - Runtime `linkSoftware` packages are retained too, so a later `mountFs` no longer unprojects them. - The Rust client `mount_fs` remains in-process-only; sidecar-visible runtime mounts there are a follow-up. --- CLAUDE.md | 7 + examples/filesystem/mount-custom-vfs.ts | 2 +- packages/core/src/agent-os.ts | 26 ++- packages/core/src/runtime-compat.ts | 13 +- packages/core/src/runtime.ts | 4 +- packages/core/src/sidecar/rpc-client.ts | 60 ++++++- packages/core/tests/mount-reconfigure.test.ts | 154 ++++++++++++++++++ packages/core/tests/mount.test.ts | 10 +- packages/runtime-core/src/kernel-proxy.ts | 4 + website/src/content/docs/docs/filesystem.mdx | 2 +- 10 files changed, 256 insertions(+), 26 deletions(-) create mode 100644 packages/core/tests/mount-reconfigure.test.ts diff --git a/CLAUDE.md b/CLAUDE.md index 018390e89..b0298183e 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -39,6 +39,13 @@ by default, warn near threshold, and fail with a typed error that names the limit and how to raise it. Host-visible warnings/errors must reach stderr/log or structured trace paths, not stay trapped in the VM. +Never swallow errors silently. Every failure must either propagate as a hard, +typed error to the caller (preferred) or be clearly logged at the failure site; +empty `catch`/`let _ =` on fallible operations and fire-and-forget promises +that drop rejections are bugs, not defensive coding. For guest-visible +surfaces, prefer matching Linux behavior — the correct POSIX errno delivered to +the guest — over inventing a softer fallback that hides the failure. + ## Runtime And Registry - The projected `/opt/agentos` filesystem is the source of truth for software diff --git a/examples/filesystem/mount-custom-vfs.ts b/examples/filesystem/mount-custom-vfs.ts index 8650f8624..df2ed11d6 100644 --- a/examples/filesystem/mount-custom-vfs.ts +++ b/examples/filesystem/mount-custom-vfs.ts @@ -3,5 +3,5 @@ import { AgentOs, createInMemoryFileSystem } from "@rivet-dev/agentos-core"; const vm = await AgentOs.create({ defaultSoftware: false }); const driver = createInMemoryFileSystem(); -vm.mountFs("/home/agentos/scratch", driver); +await vm.mountFs("/home/agentos/scratch", driver); await vm.writeFile("/home/agentos/scratch/hello.txt", "hello"); diff --git a/packages/core/src/agent-os.ts b/packages/core/src/agent-os.ts index c3ead9d99..6e1a95285 100644 --- a/packages/core/src/agent-os.ts +++ b/packages/core/src/agent-os.ts @@ -2821,6 +2821,12 @@ export class AgentOs { permissions: sidecarPermissions, commandPermissions: {}, loopbackExemptPorts: options?.loopbackExemptPorts, + // Retained for runtime mount reconfigures: `configure_vm` is + // replace-on-write for the whole payload, so post-boot mountFs + // must resend the boot packages and tool shims. + packages: sidecarPackages, + packagesMountAt: OPT_AGENTOS_ROOT, + toolShimCommands: toolBootstrapCommands, commandGuestPaths, onDispose: cleanup, // The native process is owned by the AgentOsSidecar handle and @@ -3238,18 +3244,24 @@ export class AgentOs { ); } - mountFs( + /** + * Mount a filesystem into the running VM. Resolves once the mount has been + * delivered to the native sidecar, so guest code can use it immediately + * after the returned promise settles; a delivery failure rejects instead of + * leaving the mount silently host-only. + */ + async mountFs( path: string, driver: VirtualFileSystem, options?: { readOnly?: boolean }, - ): void { + ): Promise { this._assertSafeAbsolutePath(path); - this.#kernel.mountFs(path, driver, { readOnly: options?.readOnly }); + await this.#kernel.mountFs(path, driver, { readOnly: options?.readOnly }); } - unmountFs(path: string): void { + async unmountFs(path: string): Promise { this._assertSafeAbsolutePath(path); - this.#kernel.unmountFs(path); + await this.#kernel.unmountFs(path); } async move(from: string, to: string): Promise { @@ -3541,6 +3553,10 @@ export class AgentOs { ]), ), ); + // Retain the linked package for runtime mount reconfigures: + // `configure_vm` is replace-on-write, so a later `mountFs` that + // resent only the boot packages would unproject this one. + this.#kernel.registerLinkedPackage({ path: ref.path }); } // The client parses no manifests: an `agent` block in the linked package is // picked up by the sidecar (it owns the projected `/opt/agentos` and answers diff --git a/packages/core/src/runtime-compat.ts b/packages/core/src/runtime-compat.ts index 1c1c1935a..e3d065e22 100644 --- a/packages/core/src/runtime-compat.ts +++ b/packages/core/src/runtime-compat.ts @@ -429,8 +429,8 @@ export interface Kernel extends KernelInterface { path: string, fs: VirtualFileSystem, options?: { readOnly?: boolean }, - ): void; - unmountFs(path: string): void; + ): void | Promise; + unmountFs(path: string): void | Promise; readFile(path: string): Promise; writeFile(path: string, content: string | Uint8Array): Promise; mkdir(path: string): Promise; @@ -2677,7 +2677,7 @@ class NativeKernel implements Kernel { mountPath: string, filesystem: VirtualFileSystem, options?: { readOnly?: boolean }, - ): void { + ): void | Promise { const localMount = makeLocalCompatMount({ path: mountPath, fs: filesystem, @@ -2688,15 +2688,16 @@ class NativeKernel implements Kernel { (left, right) => right.path.length - left.path.length, ); if (!this.proxy) { + // Pre-boot mounts apply during kernel initialization. return; } - this.proxy.mountFs(mountPath, filesystem, { + return this.proxy.mountFs(mountPath, filesystem, { readOnly: localMount.readOnly, sidecarMount: localMount.sidecarMount, }); } - unmountFs(mountPath: string): void { + unmountFs(mountPath: string): void | Promise { const normalized = normalizePath(mountPath); const pendingIndex = this.pendingLocalMounts.findIndex( (mount) => mount.path === normalized, @@ -2704,7 +2705,7 @@ class NativeKernel implements Kernel { if (pendingIndex >= 0) { this.pendingLocalMounts.splice(pendingIndex, 1); } - this.proxy?.unmountFs(mountPath); + return this.proxy?.unmountFs(mountPath); } async readFile(targetPath: string): Promise { diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index ae9db50ae..8f3f06548 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -333,8 +333,8 @@ export interface Kernel extends KernelInterface { path: string, fs: VirtualFileSystem, options?: { readOnly?: boolean }, - ): void; - unmountFs(path: string): void; + ): void | Promise; + unmountFs(path: string): void | Promise; readFile(path: string): Promise; writeFile(path: string, content: string | Uint8Array): Promise; mkdir(path: string): Promise; diff --git a/packages/core/src/sidecar/rpc-client.ts b/packages/core/src/sidecar/rpc-client.ts index f3ce2d4d9..8c6a225ec 100644 --- a/packages/core/src/sidecar/rpc-client.ts +++ b/packages/core/src/sidecar/rpc-client.ts @@ -339,6 +339,16 @@ interface NativeSidecarKernelProxyOptions { SidecarProcess["configureVm"] >[2]["commandPermissions"]; loopbackExemptPorts?: number[]; + /** + * The boot `configureVm` payload pieces beyond mounts/permissions. Rust + * `configure_vm` rebuilds the whole VM configuration from each payload, so + * every runtime mount reconfigure must resend these or a post-boot + * `mountFs()` silently drops the `/opt/agentos` package projections and + * tool shim commands applied at boot. + */ + packages?: Parameters[2]["packages"]; + packagesMountAt?: string; + toolShimCommands?: string[]; commandGuestPaths: ReadonlyMap; onWasmCommandResolved?: (command: string) => void; onDispose?: () => Promise; @@ -371,6 +381,13 @@ export class NativeSidecarKernelProxy { | Parameters[2]["commandPermissions"] | undefined; private readonly loopbackExemptPorts: number[] | undefined; + // Mutable: runtime `linkSoftware` appends via `registerLinkedPackage` so + // later mount reconfigures resend linked packages too, not just boot ones. + private packages: NonNullable< + Parameters[2]["packages"] + >; + private readonly packagesMountAt: string | undefined; + private readonly toolShimCommands: string[] | undefined; private readonly commandDrivers: Map; private readonly onWasmCommandResolved: | ((command: string) => void) @@ -420,6 +437,9 @@ export class NativeSidecarKernelProxy { this.permissions = options.permissions; this.commandPermissions = options.commandPermissions; this.loopbackExemptPorts = options.loopbackExemptPorts; + this.packages = options.packages ? [...options.packages] : []; + this.packagesMountAt = options.packagesMountAt; + this.toolShimCommands = options.toolShimCommands; this.commandDrivers = buildCommandMap(options.commandGuestPaths); this.onWasmCommandResolved = options.onWasmCommandResolved; this.onDispose = options.onDispose; @@ -449,6 +469,18 @@ export class NativeSidecarKernelProxy { } } + /** + * Record a runtime-linked package (`linkSoftware`) so mount reconfigures + * resend it. Rust `configure_vm` rebuilds the whole VM configuration from + * each payload, so a linked package omitted here would be silently + * unprojected from `/opt/agentos` by the next `mountFs`/`unmountFs`. + */ + registerLinkedPackage(descriptor: { path: string }): void { + if (!this.packages.some((pkg) => pkg.path === descriptor.path)) { + this.packages.push({ path: descriptor.path }); + } + } + async dispose(): Promise { if (this.disposed) { return; @@ -1506,7 +1538,7 @@ export class NativeSidecarKernelProxy { path: string, driver: VirtualFileSystem, options?: { readOnly?: boolean; sidecarMount?: SidecarMountDescriptor }, - ): void { + ): Promise { this.localMounts.unshift({ path: posixPath.normalize(path), fs: driver, @@ -1516,18 +1548,27 @@ export class NativeSidecarKernelProxy { this.localMounts.sort( (left, right) => right.path.length - left.path.length, ); - void this.reconfigureSidecarMounts().catch(() => {}); + // Resolves once the sidecar has the mount; a swallowed rejection here + // used to turn reconfigure failures into silently missing guest mounts. + // The local catch only guards callers that drop the promise — awaiting + // callers still observe the rejection. + const applied = this.reconfigureSidecarMounts(); + applied.catch(() => {}); + return applied; } - unmountFs(path: string): void { + unmountFs(path: string): Promise { const normalized = posixPath.normalize(path); const index = this.localMounts.findIndex( (mount) => mount.path === normalized, ); - if (index >= 0) { - this.localMounts.splice(index, 1); - void this.reconfigureSidecarMounts().catch(() => {}); + if (index < 0) { + return Promise.resolve(); } + this.localMounts.splice(index, 1); + const applied = this.reconfigureSidecarMounts(); + applied.catch(() => {}); + return applied; } private desiredSidecarMounts(): SidecarMountDescriptor[] { @@ -1552,11 +1593,18 @@ export class NativeSidecarKernelProxy { if (this.disposed) { return; } + // Rust `configure_vm` rebuilds the whole VM configuration from this + // payload, so resend the boot packages / tool shim commands too — + // omitting them here strips the `/opt/agentos` projections and tool + // shims from the VM as a side effect of a runtime mount change. await this.client.configureVm(this.session, this.vm, { mounts: this.desiredSidecarMounts(), permissions: this.permissions, commandPermissions: this.commandPermissions, loopbackExemptPorts: this.loopbackExemptPorts, + packages: this.packages, + packagesMountAt: this.packagesMountAt, + toolShimCommands: this.toolShimCommands, }); }; const previous = this.mountReconfigurePromise ?? Promise.resolve(); diff --git a/packages/core/tests/mount-reconfigure.test.ts b/packages/core/tests/mount-reconfigure.test.ts new file mode 100644 index 000000000..a4a366766 --- /dev/null +++ b/packages/core/tests/mount-reconfigure.test.ts @@ -0,0 +1,154 @@ +import { describe, expect, it } from "vitest"; +import { createInMemoryFileSystem } from "../src/runtime-compat.js"; +import { NativeSidecarKernelProxy } from "../src/sidecar/rpc-client.js"; + +// Regression coverage for post-boot mountFs delivery to the native sidecar: +// 1. Rust `configure_vm` rebuilds the whole VM configuration from each +// payload, so a runtime mount reconfigure that omits the boot `packages` / +// `packagesMountAt` / `toolShimCommands` strips the `/opt/agentos` +// projections and tool shims from the VM as a side effect. +// 2. mountFs used to be fire-and-forget with a swallowed rejection, so a +// failed reconfigure left the mount silently host-only and callers had no +// way to know when (or whether) the guest could see it. +// The proxy is exercised against a stub SidecarProcess so the test stays fast +// and deterministic without booting a real VM. + +const session = { connectionId: "conn-1", sessionId: "sess-1" }; +const vm = { vmId: "vm-test" }; + +const bootPackages = [ + { + name: "common", + version: "1.0.0", + path: "/tmp/common.aospkg", + }, +]; +const bootToolShims = ["agentos", "agentos-demo"]; + +function createStubClient(options?: { failConfigureVm?: boolean }) { + const configureCalls: Array> = []; + const client = { + async configureVm( + _session: unknown, + _vm: unknown, + payload: Record, + ) { + configureCalls.push(payload); + if (options?.failConfigureVm) { + throw new Error("configure_vm rejected"); + } + return { + appliedMounts: [], + appliedSoftware: [], + projectedCommands: [], + agents: [], + }; + }, + async disposeVm() {}, + async dispose() {}, + waitForEvent( + _filter: unknown, + _unused: unknown, + opts: { signal: AbortSignal }, + ) { + return new Promise((_resolve, reject) => { + opts.signal.addEventListener("abort", () => + reject(new Error("aborted")), + ); + }); + }, + }; + return { client, configureCalls }; +} + +function createProxy(client: unknown) { + const options = { + client, + session, + vm, + env: {}, + cwd: "/work", + localMounts: [], + sidecarMounts: [], + packages: bootPackages, + packagesMountAt: "/opt/agentos", + toolShimCommands: bootToolShims, + commandGuestPaths: new Map(), + ownsClient: true, + }; + return new NativeSidecarKernelProxy( + options as ConstructorParameters[0], + ); +} + +describe("post-boot mount reconfiguration", () => { + it("resends the boot packages and tool shims on runtime mountFs", async () => { + const { client, configureCalls } = createStubClient(); + const proxy = createProxy(client); + + await proxy.mountFs("/mnt/dynamic", createInMemoryFileSystem()); + + expect(configureCalls).toHaveLength(1); + const payload = configureCalls[0]; + expect(payload.packages).toEqual(bootPackages); + expect(payload.packagesMountAt).toBe("/opt/agentos"); + expect(payload.toolShimCommands).toEqual(bootToolShims); + expect(payload.mounts).toEqual([ + expect.objectContaining({ guestPath: "/mnt/dynamic" }), + ]); + + await proxy.unmountFs("/mnt/dynamic"); + expect(configureCalls).toHaveLength(2); + expect(configureCalls[1].mounts).toEqual([]); + expect(configureCalls[1].packages).toEqual(bootPackages); + expect(configureCalls[1].toolShimCommands).toEqual(bootToolShims); + + await proxy.dispose(); + }); + + it("resends runtime-linked packages on later mount reconfigures", async () => { + const { client, configureCalls } = createStubClient(); + const proxy = createProxy(client); + + // linkSoftware() records the linked package on the proxy; a later + // mountFs must resend it alongside the boot packages or configure_vm + // (replace-on-write) unprojects it from /opt/agentos. + proxy.registerLinkedPackage({ path: "/tmp/linked.aospkg" }); + await proxy.mountFs("/mnt/dynamic", createInMemoryFileSystem()); + expect(configureCalls[0].packages).toEqual([ + ...bootPackages, + { path: "/tmp/linked.aospkg" }, + ]); + + // Duplicate registration is a no-op. + proxy.registerLinkedPackage({ path: "/tmp/linked.aospkg" }); + await proxy.unmountFs("/mnt/dynamic"); + expect(configureCalls[1].packages).toEqual([ + ...bootPackages, + { path: "/tmp/linked.aospkg" }, + ]); + + await proxy.dispose(); + }); + + it("rejects the mountFs promise when sidecar delivery fails", async () => { + const { client } = createStubClient({ failConfigureVm: true }); + const proxy = createProxy(client); + + await expect( + proxy.mountFs("/mnt/dynamic", createInMemoryFileSystem()), + ).rejects.toThrow("configure_vm rejected"); + + await proxy.dispose(); + }); + + it("resolves unmountFs immediately for an unknown mount without reconfiguring", async () => { + const { client, configureCalls } = createStubClient(); + const proxy = createProxy(client); + + await proxy.unmountFs("/mnt/never-mounted"); + expect(configureCalls).toHaveLength(0); + + await proxy.dispose(); + }); +}); diff --git a/packages/core/tests/mount.test.ts b/packages/core/tests/mount.test.ts index 1e09f99e7..26880e541 100644 --- a/packages/core/tests/mount.test.ts +++ b/packages/core/tests/mount.test.ts @@ -119,12 +119,12 @@ describe("mount integration", () => { test("runtime mountFs and unmountFs work", async () => { vm = await createMountVm(); - vm.mountFs("/mnt/dynamic", createInMemoryFileSystem()); + await vm.mountFs("/mnt/dynamic", createInMemoryFileSystem()); await vm.writeFile("/mnt/dynamic/test.txt", "dynamic"); const data = await vm.readFile("/mnt/dynamic/test.txt"); expect(new TextDecoder().decode(data)).toBe("dynamic"); - vm.unmountFs("/mnt/dynamic"); + await vm.unmountFs("/mnt/dynamic"); await expect(vm.readFile("/mnt/dynamic/test.txt")).rejects.toThrow(); // Runtime mount + unmount each trigger a full sidecar reconfigure, so this // integration test needs more than the 30s default (see PR #1521 CI). @@ -134,7 +134,7 @@ describe("mount integration", () => { const mounted = createRecordingFilesystem(); vm = await createMountVm(); - vm.mountFs("/mnt/custom", mounted.fs); + await vm.mountFs("/mnt/custom", mounted.fs); await vm.writeFile("/mnt/custom/note.txt", "from custom vfs"); expect( @@ -143,7 +143,7 @@ describe("mount integration", () => { expect(mounted.calls).toContain("writeFile:/note.txt"); expect(mounted.calls).toContain("readFile:/note.txt"); - vm.unmountFs("/mnt/custom"); + await vm.unmountFs("/mnt/custom"); await expect(vm.readFile("/mnt/custom/note.txt")).rejects.toThrow(); // Runtime mount + unmount each trigger a full sidecar reconfigure, so this // integration test needs more than the 30s default (see PR #1521 CI). @@ -177,7 +177,7 @@ describe("mount integration", () => { const mounted = createRecordingFilesystem(); vm = await createMountVm(); - vm.mountFs("/mnt/custom", mounted.fs); + await vm.mountFs("/mnt/custom", mounted.fs); await vm.writeFile("/mnt/custom/host.txt", "from host api"); const result = await vm.execArgv("node", [ "-e", diff --git a/packages/runtime-core/src/kernel-proxy.ts b/packages/runtime-core/src/kernel-proxy.ts index 7eb8e55be..d41b7c0a5 100644 --- a/packages/runtime-core/src/kernel-proxy.ts +++ b/packages/runtime-core/src/kernel-proxy.ts @@ -1262,6 +1262,10 @@ export class NativeSidecarKernelProxy { return this.client.rename(this.session, this.vm, oldPath, newPath); } + // Test-runtime only: runtime mounts registered here stay host-side local + // compat mounts and are not delivered to the sidecar. The production proxy + // in @rivet-dev/agentos-core reconfigures sidecar mounts on every + // mountFs/unmountFs. mountFs( path: string, driver: VirtualFileSystem, diff --git a/website/src/content/docs/docs/filesystem.mdx b/website/src/content/docs/docs/filesystem.mdx index 6cd0b8064..491bcf8e0 100644 --- a/website/src/content/docs/docs/filesystem.mdx +++ b/website/src/content/docs/docs/filesystem.mdx @@ -50,7 +50,7 @@ Use the built-in `memory` plugin for an ephemeral mounted directory in the nativ -Use `mountFs()` for a callback-backed JS filesystem driver. The driver must live in the same JS process as the `AgentOs` instance, such as direct core usage or a custom RivetKit actor that owns an `AgentOs` instance. +Use `mountFs()` for a callback-backed JS filesystem driver. The driver must live in the same JS process as the `AgentOs` instance, such as direct core usage or a custom RivetKit actor that owns an `AgentOs` instance. `mountFs()` works on a running VM too — it returns a promise that resolves once the mount is visible to guest code, and rejects if delivery to the runtime fails. From d9179cc1b064a94f0c726cee459cb8fd29b35ea0 Mon Sep 17 00:00:00 2001 From: Nathan Flurry Date: Tue, 7 Jul 2026 19:22:18 -0700 Subject: [PATCH 4/4] feat(runtime-core): replace the sidecar frame timeout with a heartbeat silence watchdog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - The native sidecar emits a connection-scoped `StructuredEvent{name:"heartbeat"}` every 10s from a dedicated thread, so beats keep flowing while the dispatch loop is busy inside one long request. No protocol schema change — reuses the existing structured-event variant. - The host replaces the per-frame 120s timeout with a silence watchdog: any inbound frame resets a 30s clock; sustained total silence kills the sidecar and rejects in-flight requests with a typed `SidecarSilenceTimeout` (stderr tail included). - Individual requests no longer have a deadline: a legitimately long agent turn runs for minutes without teardown, while a genuinely dead or wedged sidecar is detected in ~30s instead of 2 minutes. - Removes `NATIVE_SIDECAR_FRAME_TIMEOUT_MS`, the `AGENTOS_SIDECAR_FRAME_TIMEOUT_MS` env override (#1641), and all `frameTimeoutMs` plumbing (spawn options, native-client, protocol-client, frame-rpc, correlation). - Heartbeats are swallowed at the frame layer on both clients so they never reach event consumers or the bounded event buffer; the Rust client mirrors the watchdog (activity-reset reader, kill + fail-pending on silence). - Heartbeat cadence and silence window are fixed protocol constants (no env knobs); docs updated. --- crates/native-sidecar/src/stdio.rs | 67 ++++++++ crates/sidecar-client/Cargo.toml | 2 +- crates/sidecar-client/src/transport.rs | 143 ++++++++++++++++++ packages/core/src/agent-os.ts | 2 - packages/core/src/runtime-compat.ts | 2 - .../core/src/sidecar/native-process-client.ts | 1 - packages/core/src/sidecar/rpc-client.ts | 1 - ...native-sidecar-process-permissions.test.ts | 4 - .../core/tests/native-sidecar-process.test.ts | 11 -- packages/core/tests/session-cleanup.test.ts | 2 +- packages/runtime-core/src/correlation.ts | 24 +-- packages/runtime-core/src/frame-rpc.ts | 24 ++- packages/runtime-core/src/kernel-proxy.ts | 1 - packages/runtime-core/src/native-client.ts | 27 +++- packages/runtime-core/src/protocol-client.ts | 88 ++++++++++- packages/runtime-core/src/sidecar-errors.ts | 21 +++ packages/runtime-core/src/sidecar-process.ts | 33 ++-- packages/runtime-core/src/test-runtime.ts | 2 - .../runtime-core/tests/correlation.test.ts | 62 +++----- packages/runtime-core/tests/frame-rpc.test.ts | 10 +- .../runtime-core/tests/native-client.test.ts | 2 - .../tests/protocol-client.test.ts | 120 ++++++++++++++- .../src/content/docs/docs/resource-limits.mdx | 14 ++ 23 files changed, 523 insertions(+), 140 deletions(-) diff --git a/crates/native-sidecar/src/stdio.rs b/crates/native-sidecar/src/stdio.rs index 25894a482..fb1bfe15e 100644 --- a/crates/native-sidecar/src/stdio.rs +++ b/crates/native-sidecar/src/stdio.rs @@ -47,6 +47,17 @@ use tokio::time; // pumps are cheap no-ops (try_recv + zero-timeout poll), so the higher cadence // costs negligible CPU when no guest is issuing RPCs. const EVENT_PUMP_INTERVAL: Duration = Duration::from_micros(250); +// Cadence of sidecar→host heartbeat frames. The host treats sustained inbound +// silence (several missed beats) as a dead or wedged sidecar and tears the +// process down, so this is a fixed protocol constant, not a tunable. Emitted +// from a dedicated thread so beats keep flowing while the dispatch loop is +// busy inside one long request. +const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(10); +// Connection id stamped on heartbeat frames. Heartbeats are transport-level +// liveness — not tied to an authenticated connection — and the host consumes +// them at its frame layer without routing by ownership, so a fixed synthetic +// id is correct even before any client authenticates. +const HEARTBEAT_CONNECTION_ID: &str = "sidecar-transport"; const MAX_STDIN_FRAME_QUEUE: usize = 128; const MAX_EVENT_READY_QUEUE: usize = 1; // Defense-in-depth headroom for the host-bound frame queue: a burst of output @@ -184,6 +195,7 @@ async fn run_async(extensions: Vec>) -> Result<(), Box, interval: Duration) { + thread::spawn(move || loop { + thread::sleep(interval); + let frame = match crate::service::structured_event_frame( + HEARTBEAT_CONNECTION_ID, + "heartbeat", + std::collections::HashMap::new(), + ) { + Ok(frame) => frame, + Err(error) => { + // Unreachable for a fixed name/empty detail; if it ever fires, + // stop loudly instead of spinning on a broken encoder. + tracing::error!( + target: "agentos_native_sidecar::stdio", + %error, + "failed to encode heartbeat frame; stopping heartbeat thread", + ); + return; + } + }; + if send_output_frame(&write_tx, ProtocolFrame::EventFrame(frame)).is_err() { + // Writer thread gone — the sidecar is shutting down. Normal exit. + return; + } + }); +} + fn default_compile_cache_root() -> PathBuf { // Stable across sidecar processes so V8 compile-cache (cachedData) survives a // fresh sidecar/VM and benefits cold starts. Previously keyed by PID, which @@ -707,6 +752,28 @@ mod tests { const TEST_EXTENSION_NAMESPACE: &str = "dev.rivet.secure-exec.test.blocking"; + #[test] + fn heartbeat_thread_emits_periodic_structured_heartbeat_frames() { + let (write_tx, write_rx) = + tracked_sync_channel::(TrackedLimit::SidecarStdoutFrames, 16); + spawn_heartbeat_thread(write_tx, Duration::from_millis(5)); + + // Two beats prove the emitter is periodic, not one-shot. + for beat in 0..2 { + let frame = write_rx.recv().expect("heartbeat frame"); + let ProtocolFrame::EventFrame(event) = frame else { + panic!("expected event frame for beat {beat}, got {frame:?}"); + }; + let event = crate::wire::event_frame_to_compat(event).expect("decode heartbeat frame"); + let crate::protocol::EventPayload::Structured(structured) = event.payload else { + panic!("expected structured payload for beat {beat}"); + }; + assert_eq!(structured.name, "heartbeat"); + } + // Dropping the receiver disconnects the channel; the emitter thread + // observes the send failure and exits cleanly. + } + #[test] fn read_frame_rejects_oversized_prefix_before_allocating_payload() { let codec = WireFrameCodec::new(16); diff --git a/crates/sidecar-client/Cargo.toml b/crates/sidecar-client/Cargo.toml index b70d74832..926e6b38b 100644 --- a/crates/sidecar-client/Cargo.toml +++ b/crates/sidecar-client/Cargo.toml @@ -13,5 +13,5 @@ futures = "0.3" parking_lot = "0.12" scc = "2" thiserror = "2" -tokio = { version = "1", features = ["io-util", "macros", "process", "rt", "sync"] } +tokio = { version = "1", features = ["io-util", "macros", "process", "rt", "sync", "time"] } tracing = "0.1" diff --git a/crates/sidecar-client/src/transport.rs b/crates/sidecar-client/src/transport.rs index 07c84fe26..ce6b08759 100644 --- a/crates/sidecar-client/src/transport.rs +++ b/crates/sidecar-client/src/transport.rs @@ -36,6 +36,13 @@ const PENDING_REQUEST_LIMIT: usize = 4096; /// Product clients can pass an explicit binary path to [`SidecarTransport::spawn`]. const SIDECAR_BIN_ENV: &str = "AGENTOS_SIDECAR_BIN"; +/// How long the host tolerates TOTAL inbound silence (no responses, events, sidecar requests, or +/// heartbeats) before declaring the sidecar dead. The sidecar heartbeats every 10s from a dedicated +/// thread even while busy, so this allows two missed beats plus margin; it bounds "sidecar is dead +/// or wedged", never "this request is slow" — individual requests have no deadline of their own. +/// Fixed protocol constant paired with the sidecar heartbeat cadence; mirrors the TS client. +const SIDECAR_SILENCE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); + /// A registered callback that answers a sidecar-initiated request using generated wire types. pub type WireSidecarCallback = Arc< dyn Fn( @@ -68,6 +75,8 @@ pub struct SidecarTransport { request_writer_tx: mpsc::Sender>, /// Outbound callback/control response frames. The writer drains this before regular requests. control_writer_tx: mpsc::Sender>, + /// When the reader last received any inbound frame; the silence watchdog reads it. + last_inbound_at: parking_lot::Mutex, } impl SidecarTransport { @@ -110,10 +119,15 @@ impl SidecarTransport { callbacks: SccHashMap::new(), request_writer_tx, control_writer_tx, + last_inbound_at: parking_lot::Mutex::new(std::time::Instant::now()), }); tokio::spawn(run_writer(stdin, control_writer_rx, request_writer_rx)); tokio::spawn(run_reader(Arc::downgrade(&transport), stdout)); + tokio::spawn(run_silence_watchdog( + Arc::downgrade(&transport), + SIDECAR_SILENCE_TIMEOUT, + )); Ok(transport) } @@ -236,6 +250,17 @@ impl SidecarTransport { } } wire::ProtocolFrame::EventFrame(event) => { + // Transport-level liveness beats from the sidecar. Their arrival + // already reset the silence watchdog in the reader; they carry no + // meaning for event subscribers, so drop them here (mirrors the + // TS client's heartbeat swallow). + if matches!( + &event.payload, + wire::EventPayload::StructuredEvent(structured) + if structured.name == "heartbeat" + ) { + return; + } let _ = self.event_tx.send((event.ownership, event.payload)); } wire::ProtocolFrame::SidecarRequestFrame(request) => { @@ -429,6 +454,9 @@ async fn run_reader(transport: Weak, mut stdout: ChildStdout) if stdout.read_exact(&mut frame_bytes[4..]).await.is_err() { break; } + // Any complete inbound frame proves the sidecar is alive; the silence + // watchdog measures from here. + *transport.last_inbound_at.lock() = std::time::Instant::now(); let codec = WireFrameCodec::new(max_frame_bytes); match codec.decode(&frame_bytes) { @@ -446,6 +474,30 @@ fn frame_length_exceeds_limit(length: usize, max_frame_bytes: usize) -> bool { length > max_frame_bytes } +/// Kill the sidecar and fail all in-flight requests once the transport has seen no inbound frames +/// (not even heartbeats) for `timeout`. A silent sidecar is dead or wedged, not busy: a busy +/// sidecar still heartbeats every 10s from a dedicated thread. Exits when the transport drops. +async fn run_silence_watchdog(transport: Weak, timeout: std::time::Duration) { + let check_interval = (timeout / 4).min(std::time::Duration::from_secs(1)); + loop { + tokio::time::sleep(check_interval).await; + let Some(transport) = transport.upgrade() else { + return; + }; + let silence = transport.last_inbound_at.lock().elapsed(); + if silence < timeout { + continue; + } + tracing::error!( + silence_ms = silence.as_millis() as u64, + "sidecar unresponsive: no protocol frames or heartbeats; killing sidecar", + ); + transport.kill_child(); + transport.fail_all_pending(); + return; + } +} + fn resolve_sidecar_binary_path(binary_path: Option) -> String { binary_path .or_else(|| std::env::var(SIDECAR_BIN_ENV).ok()) @@ -473,6 +525,7 @@ mod tests { callbacks: SccHashMap::new(), request_writer_tx, control_writer_tx, + last_inbound_at: parking_lot::Mutex::new(std::time::Instant::now()), } } @@ -601,6 +654,96 @@ mod tests { )); } + #[tokio::test] + async fn silence_watchdog_fails_pending_requests_after_sustained_silence() { + let transport = Arc::new(test_transport()); + let (tx, rx) = oneshot::channel(); + transport + .register_pending_request(1, tx) + .expect("register pending request"); + + tokio::spawn(run_silence_watchdog( + Arc::downgrade(&transport), + std::time::Duration::from_millis(40), + )); + + // No inbound activity at all: the watchdog must reject the pending + // request (dropped sender -> disconnected error at the caller). + rx.await + .expect_err("watchdog should drop the pending sender"); + assert_eq!(pending_request_count(&transport), 0); + } + + #[tokio::test] + async fn silence_watchdog_stays_quiet_while_frames_arrive() { + let transport = Arc::new(test_transport()); + let (tx, mut rx) = oneshot::channel(); + transport + .register_pending_request(1, tx) + .expect("register pending request"); + + tokio::spawn(run_silence_watchdog( + Arc::downgrade(&transport), + std::time::Duration::from_millis(120), + )); + + // Simulate steady inbound activity (what the reader does per frame) + // for well past the silence window; the watchdog must not fire. + for _ in 0..6 { + tokio::time::sleep(std::time::Duration::from_millis(40)).await; + *transport.last_inbound_at.lock() = std::time::Instant::now(); + assert!( + rx.try_recv().is_err(), + "pending request must remain registered while frames arrive" + ); + } + assert_eq!(pending_request_count(&transport), 1); + } + + #[tokio::test] + async fn heartbeat_events_are_swallowed_before_the_event_fanout() { + let transport = Arc::new(test_transport()); + let mut wire_events = transport.subscribe_wire_events(); + + transport + .handle_wire_frame(wire::ProtocolFrame::EventFrame(wire::EventFrame { + schema: wire::protocol_schema(), + ownership: wire::OwnershipScope::ConnectionOwnership(wire::ConnectionOwnership { + connection_id: "sidecar-transport".to_string(), + }), + payload: wire::EventPayload::StructuredEvent(wire::StructuredEvent { + name: "heartbeat".to_string(), + detail: std::collections::HashMap::new(), + }), + })) + .await; + // A non-heartbeat structured event still fans out, proving the filter + // is name-scoped rather than dropping all structured events. + transport + .handle_wire_frame(wire::ProtocolFrame::EventFrame(wire::EventFrame { + schema: wire::protocol_schema(), + ownership: wire::OwnershipScope::ConnectionOwnership(wire::ConnectionOwnership { + connection_id: "conn-1".to_string(), + }), + payload: wire::EventPayload::StructuredEvent(wire::StructuredEvent { + name: "limit_warning".to_string(), + detail: std::collections::HashMap::new(), + }), + })) + .await; + + let (_, payload) = wire_events.recv().await.expect("structured event"); + assert!(matches!( + payload, + wire::EventPayload::StructuredEvent(wire::StructuredEvent { name, .. }) + if name == "limit_warning" + )); + assert!( + wire_events.try_recv().is_err(), + "heartbeat must not fan out" + ); + } + #[test] fn pending_request_guard_removes_registered_slot_on_drop() { let transport = test_transport(); diff --git a/packages/core/src/agent-os.ts b/packages/core/src/agent-os.ts index 6e1a95285..7c8807fe1 100644 --- a/packages/core/src/agent-os.ts +++ b/packages/core/src/agent-os.ts @@ -217,7 +217,6 @@ import { type AuthenticatedSession, type CreatedVm, createAgentOsSidecarClient, - NATIVE_SIDECAR_FRAME_TIMEOUT_MS, NativeSidecarKernelProxy, type RootFilesystemEntry, type SidecarMountDescriptor, @@ -5534,7 +5533,6 @@ function ensureSharedSidecarNativeProcess( cwd: REPO_ROOT, command: ensureNativeSidecarBinary(), args: [], - frameTimeoutMs: NATIVE_SIDECAR_FRAME_TIMEOUT_MS, }); // Track the child immediately — BEFORE the handshake await — so a // failed `authenticateAndOpenSession()` can still reap it (otherwise diff --git a/packages/core/src/runtime-compat.ts b/packages/core/src/runtime-compat.ts index e3d065e22..7dd2dd45f 100644 --- a/packages/core/src/runtime-compat.ts +++ b/packages/core/src/runtime-compat.ts @@ -15,7 +15,6 @@ import { type AuthenticatedSession, type CreatedVm, type LocalCompatMount, - NATIVE_SIDECAR_FRAME_TIMEOUT_MS, NativeSidecarKernelProxy, SidecarProcess, type RootFilesystemEntry, @@ -2846,7 +2845,6 @@ class NativeKernel implements Kernel { cwd: REPO_ROOT, command: ensureNativeSidecarBinary(), args: [], - frameTimeoutMs: NATIVE_SIDECAR_FRAME_TIMEOUT_MS, }); const session = await client.authenticateAndOpenSession(); const createVmConfig: CreateVmConfig = { diff --git a/packages/core/src/sidecar/native-process-client.ts b/packages/core/src/sidecar/native-process-client.ts index 37d6cf149..68d99e99d 100644 --- a/packages/core/src/sidecar/native-process-client.ts +++ b/packages/core/src/sidecar/native-process-client.ts @@ -6,7 +6,6 @@ import "@rivet-dev/agentos-runtime-core/native-client"; import { SidecarProcess } from "@rivet-dev/agentos-runtime-core/sidecar-client"; export { - NATIVE_SIDECAR_FRAME_TIMEOUT_MS, SidecarEventBufferOverflow, SidecarProcess, SidecarProcessError, diff --git a/packages/core/src/sidecar/rpc-client.ts b/packages/core/src/sidecar/rpc-client.ts index 8c6a225ec..2ed242f95 100644 --- a/packages/core/src/sidecar/rpc-client.ts +++ b/packages/core/src/sidecar/rpc-client.ts @@ -2755,7 +2755,6 @@ export type { SidecarSpawnOptions, } from "./native-process-client.js"; export { - NATIVE_SIDECAR_FRAME_TIMEOUT_MS, NativeSidecarProcessClient, SidecarEventBufferOverflow, SidecarProcess, diff --git a/packages/core/tests/native-sidecar-process-permissions.test.ts b/packages/core/tests/native-sidecar-process-permissions.test.ts index 867a8bace..f0246f6cd 100644 --- a/packages/core/tests/native-sidecar-process-permissions.test.ts +++ b/packages/core/tests/native-sidecar-process-permissions.test.ts @@ -193,7 +193,6 @@ describe("native sidecar process client permissions", () => { cwd: REPO_ROOT, command: "node", args: [driverPath, capturePath], - frameTimeoutMs: 5_000, payloadCodec: "json", }); @@ -306,7 +305,6 @@ describe("native sidecar process client permissions", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { @@ -394,7 +392,6 @@ describe("native sidecar process client permissions", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { @@ -631,7 +628,6 @@ describe("native sidecar process client permissions", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { diff --git a/packages/core/tests/native-sidecar-process.test.ts b/packages/core/tests/native-sidecar-process.test.ts index eedf40478..663900892 100644 --- a/packages/core/tests/native-sidecar-process.test.ts +++ b/packages/core/tests/native-sidecar-process.test.ts @@ -458,7 +458,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: "node", args: [driverPath, capturePath], - frameTimeoutMs: 5_000, }); client.setSidecarRequestHandler(async (request) => { expect(request.request_id).toBe(-1); @@ -528,7 +527,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: "node", args: [driverPath], - frameTimeoutMs: 5_000, }); const childPid = ( client as unknown as { @@ -610,7 +608,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: "node", args: [driverPath], - frameTimeoutMs: 5_000, payloadCodec: "json", eventBufferCapacity: 128, }); @@ -726,7 +723,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: "node", args: [driverPath], - frameTimeoutMs: 30_000, payloadCodec: "json", }); @@ -795,7 +791,6 @@ describe("native sidecar process client", () => { `agentos-sidecar-missing-${process.pid}-${Date.now()}`, ), args: [], - frameTimeoutMs: 30_000, }); await expect(client.authenticateAndOpenSession()).rejects.toBeInstanceOf( @@ -1052,7 +1047,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { @@ -1191,7 +1185,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { @@ -1309,7 +1302,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { @@ -1444,7 +1436,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { @@ -1638,7 +1629,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { @@ -1713,7 +1703,6 @@ describe("native sidecar process client", () => { cwd: REPO_ROOT, command: SIDECAR_BINARY, args: [], - frameTimeoutMs: 20_000, }); try { diff --git a/packages/core/tests/session-cleanup.test.ts b/packages/core/tests/session-cleanup.test.ts index 5f03813a1..21f0bc58a 100644 --- a/packages/core/tests/session-cleanup.test.ts +++ b/packages/core/tests/session-cleanup.test.ts @@ -533,7 +533,7 @@ function isSharedRuntimeCloseRaceError(error: unknown): boolean { return [ "sidecar stdout closed while reading frame", "Broken pipe (os error 32)", - "timed out waiting for sidecar protocol frame for close_agent_session", + "sidecar unresponsive: no protocol frames or heartbeats", ].some((fragment) => error.message.includes(fragment)); } diff --git a/packages/runtime-core/src/correlation.ts b/packages/runtime-core/src/correlation.ts index 13f8a4ee4..981e1afea 100644 --- a/packages/runtime-core/src/correlation.ts +++ b/packages/runtime-core/src/correlation.ts @@ -1,42 +1,32 @@ interface PendingResponse { resolve: (frame: TResponse) => void; reject: (error: Error) => void; - timer: ReturnType; } export class PendingResponseRegistry { private readonly pending = new Map>(); - waitForResponse( - requestId: number, - options: { - timeoutMs: number; - timeoutMessage: () => string; - }, - ): Promise { + // Deliberately no per-request timer: local framed stdio never loses frames, + // so a response is bounded by the transport's silence watchdog (a dead or + // wedged sidecar rejects all pending requests through `rejectAll`) rather + // than by guessing how long any one request should take. + waitForResponse(requestId: number): Promise { if (this.pending.has(requestId)) { throw new Error( `response waiter already registered for request ${requestId}`, ); } return new Promise((resolve, reject) => { - const entry = { + this.pending.set(requestId, { resolve: (frame: TResponse) => { - clearTimeout(entry.timer); this.pending.delete(requestId); resolve(frame); }, reject: (error: Error) => { - clearTimeout(entry.timer); this.pending.delete(requestId); reject(error); }, - timer: setTimeout(() => { - this.pending.delete(requestId); - reject(new Error(options.timeoutMessage())); - }, options.timeoutMs), - }; - this.pending.set(requestId, entry); + }); }); } diff --git a/packages/runtime-core/src/frame-rpc.ts b/packages/runtime-core/src/frame-rpc.ts index 2e77ea540..2a7c68d6f 100644 --- a/packages/runtime-core/src/frame-rpc.ts +++ b/packages/runtime-core/src/frame-rpc.ts @@ -51,6 +51,7 @@ export class FrameRpcTransport< private readonly sidecarRequestListeners = new Set< (request: TSidecarRequestFrame) => void >(); + private readonly frameActivityListeners = new Set<() => void>(); constructor( options: FrameRpcTransportOptions< @@ -88,6 +89,19 @@ export class FrameRpcTransport< }; } + /** + * Observe every classified inbound frame (response, event, or sidecar + * request) before it is routed. This is the transport's liveness signal: + * the silence watchdog resets on each invocation, so ANY inbound traffic — + * not just heartbeats — proves the sidecar is alive. + */ + onFrameActivity(handler: () => void): () => void { + this.frameActivityListeners.add(handler); + return () => { + this.frameActivityListeners.delete(handler); + }; + } + onSidecarRequest(handler: (request: TSidecarRequestFrame) => void): () => void { this.sidecarRequestListeners.add(handler); return () => { @@ -106,12 +120,8 @@ export class FrameRpcTransport< async sendFrame( requestId: number, frame: TWriteFrame, - options: { - timeoutMs: number; - timeoutMessage: () => string; - }, ): Promise { - const response = this.pendingResponses.waitForResponse(requestId, options); + const response = this.pendingResponses.waitForResponse(requestId); void this.writeFrame(frame).catch((error) => { this.pendingResponses.reject( requestId, @@ -134,6 +144,7 @@ export class FrameRpcTransport< this.pendingResponses.rejectAll(new Error("frame rpc transport disposed")); this.eventListeners.clear(); this.sidecarRequestListeners.clear(); + this.frameActivityListeners.clear(); } private dispatchFrame( @@ -143,6 +154,9 @@ export class FrameRpcTransport< TSidecarRequestFrame >, ): void { + for (const listener of this.frameActivityListeners) { + listener(); + } switch (classified.kind) { case "response": this.pendingResponses.resolve( diff --git a/packages/runtime-core/src/kernel-proxy.ts b/packages/runtime-core/src/kernel-proxy.ts index d41b7c0a5..2a0c5e30a 100644 --- a/packages/runtime-core/src/kernel-proxy.ts +++ b/packages/runtime-core/src/kernel-proxy.ts @@ -2354,7 +2354,6 @@ export type { SidecarSocketStateEntry, } from "./sidecar-process.js"; export { - NATIVE_SIDECAR_FRAME_TIMEOUT_MS, SidecarProcess, SidecarEventBufferOverflow, SidecarProcessError, diff --git a/packages/runtime-core/src/native-client.ts b/packages/runtime-core/src/native-client.ts index 3bda2d269..3973c521f 100644 --- a/packages/runtime-core/src/native-client.ts +++ b/packages/runtime-core/src/native-client.ts @@ -16,7 +16,6 @@ import type { LiveOwnershipScope } from "./ownership.js"; import type { LiveRequestPayload } from "./request-payloads.js"; import type { LiveSidecarEventSelector } from "./event-buffer.js"; -export const DEFAULT_SIDECAR_FRAME_TIMEOUT_MS = 120_000; export const DEFAULT_SIDECAR_EVENT_BUFFER_CAPACITY = 4_096; export const DEFAULT_SIDECAR_GRACEFUL_EXIT_MS = 5_000; export const DEFAULT_SIDECAR_FORCE_EXIT_MS = 2_000; @@ -25,25 +24,30 @@ export interface StdioSidecarProtocolClientSpawnOptions { cwd?: string; command?: string; args?: string[]; - frameTimeoutMs?: number; eventBufferCapacity?: number; gracefulExitMs?: number; forceExitMs?: number; disposedErrorMessage?: string; payloadCodec?: ProtocolFramePayloadCodec; + /** + * Override the silence watchdog window (default 30s). Tests only — the + * window is a fixed protocol constant paired with the sidecar's 10s + * heartbeat cadence, not an operator tunable. + */ + silenceTimeoutMs?: number; } type ResolvedStdioSidecarProtocolClientOptions = Required< Pick< StdioSidecarProtocolClientSpawnOptions, - | "frameTimeoutMs" | "eventBufferCapacity" | "gracefulExitMs" | "forceExitMs" | "disposedErrorMessage" | "payloadCodec" > ->; +> & + Pick; export class StdioSidecarProtocolClient implements SidecarProcessTransport { readonly child: StdioSidecarProcess["child"]; @@ -65,9 +69,19 @@ export class StdioSidecarProtocolClient implements SidecarProcessTransport { this.protocolClient = new SidecarProtocolClient({ stdin: this.child.stdin, stdout: this.child.stdout, - frameTimeoutMs: options.frameTimeoutMs, eventBufferCapacity: options.eventBufferCapacity, payloadCodec: options.payloadCodec, + silenceTimeoutMs: options.silenceTimeoutMs, + // A silent sidecar is dead or wedged; reap the process so it cannot + // linger as a zombie holding VM resources. The watchdog then rejects + // all in-flight requests with `SidecarSilenceTimeout`. + onSilenceExpired: () => { + try { + this.child.kill("SIGKILL"); + } catch { + // The child may have exited between the check and the kill. + } + }, stderrText: () => this.sidecarProcess.stderrText(), streamEndedError: () => this.sidecarProcess.currentExitError() ?? @@ -96,8 +110,7 @@ export class StdioSidecarProtocolClient implements SidecarProcessTransport { cwd: options.cwd, }), { - frameTimeoutMs: - options.frameTimeoutMs ?? DEFAULT_SIDECAR_FRAME_TIMEOUT_MS, + silenceTimeoutMs: options.silenceTimeoutMs, eventBufferCapacity: options.eventBufferCapacity ?? DEFAULT_SIDECAR_EVENT_BUFFER_CAPACITY, diff --git a/packages/runtime-core/src/protocol-client.ts b/packages/runtime-core/src/protocol-client.ts index 618e2c848..caf98c72d 100644 --- a/packages/runtime-core/src/protocol-client.ts +++ b/packages/runtime-core/src/protocol-client.ts @@ -24,6 +24,16 @@ import { } from "./protocol-frames.js"; import type { LiveOwnershipScope } from "./ownership.js"; import type { LiveRequestPayload } from "./request-payloads.js"; +import { SidecarSilenceTimeout } from "./sidecar-errors.js"; + +/** + * How long the host tolerates TOTAL inbound silence (no responses, events, + * sidecar requests, or heartbeats) before declaring the sidecar dead. The + * sidecar heartbeats every 10s from a dedicated thread, so this allows two + * missed beats plus margin; it bounds "sidecar is dead or wedged", never "this + * request is slow" — individual requests have no deadline of their own. + */ +const DEFAULT_SIDECAR_SILENCE_TIMEOUT_MS = 30_000; export interface SidecarProtocolClientOptions { frameTransport?: FrameTransport< @@ -32,18 +42,26 @@ export interface SidecarProtocolClientOptions { >; stdin?: Writable; stdout?: Readable; - frameTimeoutMs: number; eventBufferCapacity: number; payloadCodec?: ProtocolFramePayloadCodec; stderrText?: () => string; frameError?: (error: Error) => Error; streamEndedError?: () => Error; + /** Override the silence watchdog window. Tests only; production uses the default. */ + silenceTimeoutMs?: number; + /** + * Runs when the silence watchdog fires, before pending work is rejected. + * The stdio layer uses it to SIGKILL the sidecar child. + */ + onSilenceExpired?: () => void; } export class SidecarProtocolClient { private readonly eventBuffer: SidecarEventBuffer; private readonly eventListeners = new Set<(event: LiveEventFrame) => void>(); - private readonly frameTimeoutMs: number; + private readonly silenceTimeoutMs: number; + private silenceTimer: ReturnType | null = null; + private lastInboundAtMs = 0; private readonly payloadCodec: ProtocolFramePayloadCodec; private readonly stderrText: () => string; private readonly hostFrameFactory = new HostProtocolFrameFactory(); @@ -64,7 +82,8 @@ export class SidecarProtocolClient { private sidecarRequestHandler: LiveSidecarRequestHandler | null = null; constructor(options: SidecarProtocolClientOptions) { - this.frameTimeoutMs = options.frameTimeoutMs; + this.silenceTimeoutMs = + options.silenceTimeoutMs ?? DEFAULT_SIDECAR_SILENCE_TIMEOUT_MS; this.eventBuffer = new SidecarEventBuffer(options.eventBufferCapacity); this.payloadCodec = options.payloadCodec ?? "bare"; this.stderrText = options.stderrText ?? (() => ""); @@ -99,6 +118,49 @@ export class SidecarProtocolClient { new Error("sidecar protocol stream ended"), ); }); + this.frameTransport.onFrameActivity(() => { + this.lastInboundAtMs = performance.now(); + }); + this.startSilenceWatchdog(options.onSilenceExpired); + } + + /** + * Arm the silence watchdog: ANY inbound frame resets the clock (see the + * `onFrameActivity` tap above), and the sidecar heartbeats every 10s even + * while busy, so sustained silence for the full window means the process + * is dead or wedged — not slow. The check interval is unref'd so an idle + * host process can still exit naturally. + */ + private startSilenceWatchdog(onExpired?: () => void): void { + this.lastInboundAtMs = performance.now(); + const checkIntervalMs = Math.max( + Math.min(this.silenceTimeoutMs / 4, 1_000), + 10, + ); + this.silenceTimer = setInterval(() => { + const silenceMs = performance.now() - this.lastInboundAtMs; + if (silenceMs < this.silenceTimeoutMs) { + return; + } + this.stopSilenceWatchdog(); + const error = new SidecarSilenceTimeout({ + silenceMs, + stderr: this.stderrText(), + }); + try { + onExpired?.(); + } finally { + this.failPermanently(error); + } + }, checkIntervalMs); + this.silenceTimer.unref?.(); + } + + private stopSilenceWatchdog(): void { + if (this.silenceTimer !== null) { + clearInterval(this.silenceTimer); + this.silenceTimer = null; + } } setSidecarRequestHandler(handler: LiveSidecarRequestHandler | null): void { @@ -121,14 +183,12 @@ export class SidecarProtocolClient { } const request = this.hostFrameFactory.createRequestFrame(input); + // No per-request deadline: only the caller knows whether an operation is + // legitimately long (a whole agent turn is one request). A dead or + // wedged sidecar rejects this via the silence watchdog instead. const response = await this.frameTransport.sendFrame( request.request_id, request, - { - timeoutMs: this.frameTimeoutMs, - timeoutMessage: () => - `timed out waiting for sidecar protocol frame for ${input.payload.type}\nstderr:\n${this.stderrText()}`, - }, ); if (response.payload.type === "rejected") { @@ -224,10 +284,12 @@ export class SidecarProtocolClient { } } this.closedError = error; + this.stopSilenceWatchdog(); this.rejectPending(error); } dispose(): void { + this.stopSilenceWatchdog(); this.frameTransport.dispose(); } @@ -258,6 +320,16 @@ export class SidecarProtocolClient { } private dispatchEvent(event: LiveEventFrame): void { + // Transport-level liveness beats from the sidecar. Their arrival already + // reset the silence watchdog at the frame layer; they carry no meaning + // for consumers and must never reach the bounded event buffer, where a + // long-idle VM would accumulate one every 10s until overflow. + if ( + event.payload.type === "structured" && + event.payload.name === "heartbeat" + ) { + return; + } for (const listener of this.eventListeners) { try { listener(event); diff --git a/packages/runtime-core/src/sidecar-errors.ts b/packages/runtime-core/src/sidecar-errors.ts index 4e43177be..ecc678eba 100644 --- a/packages/runtime-core/src/sidecar-errors.ts +++ b/packages/runtime-core/src/sidecar-errors.ts @@ -28,6 +28,27 @@ export class SidecarProcessExited extends Error { } } +/** + * The silence watchdog fired: the sidecar produced no protocol frames at all — + * not even its 10s liveness heartbeats — for the full silence window, so the + * process is dead or wedged (not merely busy: a busy sidecar still heartbeats + * from a dedicated thread). The host kills the sidecar and rejects every + * in-flight request with this error. + */ +export class SidecarSilenceTimeout extends Error { + readonly silenceMs: number; + readonly stderr: string; + + constructor(options: { silenceMs: number; stderr: string }) { + super( + `sidecar unresponsive: no protocol frames or heartbeats for ${Math.round(options.silenceMs)}ms; killing sidecar${formatSidecarStderrSuffix(options.stderr)}`, + ); + this.name = "SidecarSilenceTimeout"; + this.silenceMs = options.silenceMs; + this.stderr = options.stderr; + } +} + export class SidecarProcessError extends Error { readonly childError: Error; readonly stderr: string; diff --git a/packages/runtime-core/src/sidecar-process.ts b/packages/runtime-core/src/sidecar-process.ts index 5e10425a1..9a5e92694 100644 --- a/packages/runtime-core/src/sidecar-process.ts +++ b/packages/runtime-core/src/sidecar-process.ts @@ -47,6 +47,7 @@ import { export { SidecarProcessError, SidecarProcessExited, + SidecarSilenceTimeout, } from "./sidecar-errors.js"; export { SidecarEventBufferOverflow } from "./event-buffer.js"; // `Sidecar` is the public name for the native sidecar process client. The class @@ -56,27 +57,6 @@ export { SidecarProcess as Sidecar }; const BRIDGE_CONTRACT_VERSION = 1; -const DEFAULT_NATIVE_SIDECAR_FRAME_TIMEOUT_MS = 120_000; - -// Per-frame timeout for the native sidecar protocol: how long the host waits for -// a sidecar response frame before treating the sidecar as unresponsive. A single -// long host request maps to one frame, so workloads with legitimately long turns -// (an agent turn with many tool calls, media generation, etc.) can exceed the -// default and be killed mid-turn with "timed out waiting for sidecar protocol -// frame". The native sidecar is process-global (spawned once and multiplexed), -// so this is an env override rather than a per-instance option. -function resolveNativeSidecarFrameTimeoutMs(): number { - const raw = process.env.AGENTOS_SIDECAR_FRAME_TIMEOUT_MS; - if (raw !== undefined) { - const parsed = Number(raw); - if (Number.isFinite(parsed) && parsed > 0) { - return parsed; - } - } - return DEFAULT_NATIVE_SIDECAR_FRAME_TIMEOUT_MS; -} - -export const NATIVE_SIDECAR_FRAME_TIMEOUT_MS = resolveNativeSidecarFrameTimeoutMs(); const DEFAULT_SIDECAR_EVENT_BUFFER_CAPACITY = 4_096; const DEFAULT_SIDECAR_GRACEFUL_EXIT_MS = 5_000; const DEFAULT_SIDECAR_FORCE_EXIT_MS = 2_000; @@ -215,22 +195,27 @@ export interface SidecarSpawnOptions { cwd?: string; command?: string; args?: string[]; - frameTimeoutMs?: number; eventBufferCapacity?: number; // Migration-only compatibility path for pre-BARE test fixtures. payloadCodec?: NativeTransportPayloadCodec; + /** + * Override the sidecar silence watchdog window (default 30s). Tests only — + * it is a fixed protocol constant paired with the sidecar's 10s heartbeat + * cadence, not an operator tunable. + */ + silenceTimeoutMs?: number; } export interface ResolvedSidecarSpawnOptions { cwd?: string; command?: string; args: string[]; - frameTimeoutMs: number; eventBufferCapacity: number; gracefulExitMs: number; forceExitMs: number; disposedErrorMessage: string; payloadCodec: NativeTransportPayloadCodec; + silenceTimeoutMs?: number; } type SidecarProcessSpawnFactory = ( @@ -380,7 +365,7 @@ export class SidecarProcess { command: options.command, args: options.args ?? [], cwd: options.cwd, - frameTimeoutMs: options.frameTimeoutMs ?? NATIVE_SIDECAR_FRAME_TIMEOUT_MS, + silenceTimeoutMs: options.silenceTimeoutMs, eventBufferCapacity: options.eventBufferCapacity ?? DEFAULT_SIDECAR_EVENT_BUFFER_CAPACITY, gracefulExitMs: DEFAULT_SIDECAR_GRACEFUL_EXIT_MS, diff --git a/packages/runtime-core/src/test-runtime.ts b/packages/runtime-core/src/test-runtime.ts index 959fdd368..aeb8c1326 100644 --- a/packages/runtime-core/src/test-runtime.ts +++ b/packages/runtime-core/src/test-runtime.ts @@ -10,7 +10,6 @@ import { type AuthenticatedSession, type CreatedVm, type LocalCompatMount, - NATIVE_SIDECAR_FRAME_TIMEOUT_MS, NativeSidecarKernelProxy, SidecarProcess, type RootFilesystemEntry, @@ -3070,7 +3069,6 @@ class NativeKernel implements Kernel { cwd: REPO_ROOT, command: ensureNativeSidecarBinary(), args: [], - frameTimeoutMs: NATIVE_SIDECAR_FRAME_TIMEOUT_MS, }), ); const session = await this.measureBoot("session_open", () => diff --git a/packages/runtime-core/tests/correlation.test.ts b/packages/runtime-core/tests/correlation.test.ts index 12b84ee94..7ea101ca1 100644 --- a/packages/runtime-core/tests/correlation.test.ts +++ b/packages/runtime-core/tests/correlation.test.ts @@ -1,13 +1,10 @@ -import { describe, expect, test, vi } from "vitest"; +import { describe, expect, test } from "vitest"; import { PendingResponseRegistry } from "../src/correlation.js"; describe("pending response registry", () => { test("resolves a registered response by request id", async () => { const registry = new PendingResponseRegistry(); - const response = registry.waitForResponse(7, { - timeoutMs: 1_000, - timeoutMessage: () => "timed out", - }); + const response = registry.waitForResponse(7); expect(registry.resolve(7, "ok")).toBe(true); await expect(response).resolves.toBe("ok"); @@ -16,10 +13,7 @@ describe("pending response registry", () => { test("rejects a registered response by request id", async () => { const registry = new PendingResponseRegistry(); - const response = registry.waitForResponse(9, { - timeoutMs: 1_000, - timeoutMessage: () => "timed out", - }); + const response = registry.waitForResponse(9); const error = new Error("write failed"); expect(registry.reject(9, error)).toBe(true); @@ -27,54 +21,34 @@ describe("pending response registry", () => { expect(registry.reject(9, error)).toBe(false); }); - test("times out pending responses", async () => { - vi.useFakeTimers(); - try { - const registry = new PendingResponseRegistry(); - const response = registry.waitForResponse(11, { - timeoutMs: 25, - timeoutMessage: () => "request timed out", - }); - const expectedRejection = expect(response).rejects.toThrow( - "request timed out", - ); + test("pending responses have no deadline of their own", async () => { + // A response is bounded by the transport silence watchdog (via + // `rejectAll`), never by a per-request timer: a legitimately long + // request must be able to outlive any fixed per-request window. + const registry = new PendingResponseRegistry(); + const response = registry.waitForResponse(11); - await vi.advanceTimersByTimeAsync(25); - await expectedRejection; - expect(registry.resolve(11, "late")).toBe(false); - } finally { - vi.useRealTimers(); - } + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(registry.resolve(11, "slow but fine")).toBe(true); + await expect(response).resolves.toBe("slow but fine"); }); test("rejects duplicate request ids", () => { const registry = new PendingResponseRegistry(); - const pending = registry.waitForResponse(13, { - timeoutMs: 1_000, - timeoutMessage: () => "timed out", - }); + const pending = registry.waitForResponse(13); void pending.catch(() => undefined); - expect(() => - registry.waitForResponse(13, { - timeoutMs: 1_000, - timeoutMessage: () => "timed out", - }), - ).toThrow("response waiter already registered for request 13"); + expect(() => registry.waitForResponse(13)).toThrow( + "response waiter already registered for request 13", + ); registry.rejectAll(new Error("cleanup")); }); test("rejects all pending responses", async () => { const registry = new PendingResponseRegistry(); - const first = registry.waitForResponse(1, { - timeoutMs: 1_000, - timeoutMessage: () => "first timed out", - }); - const second = registry.waitForResponse(2, { - timeoutMs: 1_000, - timeoutMessage: () => "second timed out", - }); + const first = registry.waitForResponse(1); + const second = registry.waitForResponse(2); registry.rejectAll(new Error("transport closed")); diff --git a/packages/runtime-core/tests/frame-rpc.test.ts b/packages/runtime-core/tests/frame-rpc.test.ts index 39e3a96c9..af3ec980d 100644 --- a/packages/runtime-core/tests/frame-rpc.test.ts +++ b/packages/runtime-core/tests/frame-rpc.test.ts @@ -65,11 +65,11 @@ describe("frame RPC transport", () => { }); }); - const response = transport.sendFrame( - 3, - { frame_type: "request", request_id: 3, payload: "ping" }, - { timeoutMs: 1_000, timeoutMessage: () => "timed out" }, - ); + const response = transport.sendFrame(3, { + frame_type: "request", + request_id: 3, + payload: "ping", + }); await expect(written).resolves.toEqual({ frame_type: "request", request_id: 3, diff --git a/packages/runtime-core/tests/native-client.test.ts b/packages/runtime-core/tests/native-client.test.ts index 5a7081410..8d7404731 100644 --- a/packages/runtime-core/tests/native-client.test.ts +++ b/packages/runtime-core/tests/native-client.test.ts @@ -56,7 +56,6 @@ describe("stdio sidecar protocol client", () => { const client = StdioSidecarProtocolClient.spawn({ command: process.execPath, args: [driverPath], - frameTimeoutMs: 1_000, eventBufferCapacity: 8, payloadCodec: "json", }); @@ -117,7 +116,6 @@ describe("stdio sidecar protocol client", () => { const sidecarProcess = SidecarProcess.spawn({ command: process.execPath, args: [driverPath], - frameTimeoutMs: 1_000, eventBufferCapacity: 8, payloadCodec: "json", }); diff --git a/packages/runtime-core/tests/protocol-client.test.ts b/packages/runtime-core/tests/protocol-client.test.ts index cc9908f0c..41248089f 100644 --- a/packages/runtime-core/tests/protocol-client.test.ts +++ b/packages/runtime-core/tests/protocol-client.test.ts @@ -23,7 +23,6 @@ function createClient() { const client = new SidecarProtocolClient({ stdin, stdout, - frameTimeoutMs: 1_000, eventBufferCapacity: 8, payloadCodec: "json", stderrText: () => "stderr", @@ -138,7 +137,6 @@ describe("sidecar protocol client", () => { const frameTransport = new MemoryFrameTransport(); const client = new SidecarProtocolClient({ frameTransport, - frameTimeoutMs: 1_000, eventBufferCapacity: 8, payloadCodec: "json", stderrText: () => "stderr", @@ -201,6 +199,124 @@ describe("sidecar protocol client", () => { client.dispose(); }); + it("swallows heartbeat events before waiters and the bounded buffer", async () => { + const frameTransport = new MemoryFrameTransport(); + const client = new SidecarProtocolClient({ + frameTransport, + eventBufferCapacity: 2, + payloadCodec: "json", + stderrText: () => "stderr", + }); + const heartbeat = (): LiveEventFrame => ({ + frame_type: "event", + schema: SIDECAR_PROTOCOL_SCHEMA, + ownership, + payload: { type: "structured", name: "heartbeat", detail: {} }, + }); + + // Far more heartbeats than the buffer capacity: if any were buffered, + // the overflow would fail the client permanently and the later request + // below would reject. + for (let i = 0; i < 8; i += 1) { + frameTransport.emitFrame(heartbeat()); + } + + // A predicate waiter that would match anything must not see heartbeats. + let sawHeartbeat = false; + const waiter = client.waitForEvent((event) => { + if ( + event.payload.type === "structured" && + event.payload.name === "heartbeat" + ) { + sawHeartbeat = true; + } + return event.payload.type === "vm_lifecycle"; + }); + frameTransport.emitFrame(heartbeat()); + frameTransport.emitFrame({ + frame_type: "event", + schema: SIDECAR_PROTOCOL_SCHEMA, + ownership, + payload: { type: "vm_lifecycle", state: "ready" }, + }); + + await expect(waiter).resolves.toMatchObject({ + payload: { type: "vm_lifecycle", state: "ready" }, + }); + expect(sawHeartbeat).toBe(false); + client.dispose(); + }); + + it("rejects in-flight requests when the silence watchdog fires", async () => { + const frameTransport = new MemoryFrameTransport(); + let expired = 0; + const client = new SidecarProtocolClient({ + frameTransport, + eventBufferCapacity: 8, + payloadCodec: "json", + stderrText: () => "sidecar stderr tail", + silenceTimeoutMs: 50, + onSilenceExpired: () => { + expired += 1; + }, + }); + + const response = client.sendRequest({ + ownership, + payload: { type: "create_layer" }, + }); + + await expect(response).rejects.toThrow( + /sidecar unresponsive: no protocol frames or heartbeats for \d+ms/, + ); + expect(expired).toBe(1); + // The client is failed permanently: later requests reject immediately. + await expect( + client.sendRequest({ ownership, payload: { type: "create_layer" } }), + ).rejects.toThrow(/sidecar unresponsive/); + client.dispose(); + }); + + it("inbound frames reset the silence watchdog", async () => { + const frameTransport = new MemoryFrameTransport(); + const client = new SidecarProtocolClient({ + frameTransport, + eventBufferCapacity: 8, + payloadCodec: "json", + stderrText: () => "stderr", + silenceTimeoutMs: 120, + }); + + // Keep beating for well past the silence window; the client must stay + // healthy because every heartbeat resets the clock. + for (let i = 0; i < 6; i += 1) { + await new Promise((resolve) => setTimeout(resolve, 40)); + frameTransport.emitFrame({ + frame_type: "event", + schema: SIDECAR_PROTOCOL_SCHEMA, + ownership, + payload: { type: "structured", name: "heartbeat", detail: {} }, + }); + } + + const response = client.sendRequest({ + ownership, + payload: { type: "create_layer" }, + }); + await expect.poll(() => frameTransport.writes.length).toBe(1); + frameTransport.emitFrame({ + frame_type: "response", + schema: SIDECAR_PROTOCOL_SCHEMA, + request_id: 1, + ownership, + payload: { type: "layer_created", layer_id: "layer" }, + }); + await expect(response).resolves.toMatchObject({ + payload: { type: "layer_created", layer_id: "layer" }, + }); + client.dispose(); + }); + it("writes sidecar request handler responses", async () => { const { stdin, stdout, client } = createClient(); const written = readWrittenFrame(stdin); diff --git a/website/src/content/docs/docs/resource-limits.mdx b/website/src/content/docs/docs/resource-limits.mdx index dc3e73fef..e6c2b1314 100644 --- a/website/src/content/docs/docs/resource-limits.mdx +++ b/website/src/content/docs/docs/resource-limits.mdx @@ -46,6 +46,20 @@ Set caps on the `limits` object in the `agentOS` config. Limits are grouped by s - **Filesystem bytes**: writing past the VFS budget fails with a no-space error to the guest. - **Counts (fds / processes / sockets)**: hitting a table cap returns the standard POSIX errno (`EMFILE`, `EAGAIN`, etc.), exactly as a real Linux kernel would under `ulimit`. +## Sidecar liveness + +Separate from the guest caps above, the host detects a dead or wedged sidecar +process by silence, not by per-request deadlines. The sidecar emits a liveness +heartbeat every 10 seconds from a dedicated thread — so it keeps beating even +mid-way through a long turn — and the host treats 30 seconds with no inbound +frames at all as a dead sidecar: it kills the process and fails in-flight +requests with a typed `SidecarSilenceTimeout` error. + +Because liveness is silence-based, individual requests have no time limit of +their own: an agent turn may legitimately run for many minutes without being +torn down. Neither the heartbeat cadence nor the silence window is +configurable — they are fixed protocol constants. + ## Warnings & observability Limits are observable, not just enforced. Every bound — resource caps and the