feat(runtime-core): replace the sidecar frame timeout with a heartbeat silence watchdog

- 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.

Author: NathanFlurry

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<Box<dyn Extension>>) -> Result<(), Box<dyn Er
             }
         }
     });
+    spawn_heartbeat_thread(write_tx.clone(), HEARTBEAT_INTERVAL);
 
     thread::spawn({
         let callback_transport = callback_transport.clone();
@@ -688,6 +700,39 @@ fn send_output_frame(
     })
 }
 
+/// Emit a connection-scoped `StructuredEvent { name: "heartbeat" }` frame every
+/// `interval` for as long as the stdout writer is alive. This is the host's
+/// liveness signal: it resets the host's silence watchdog, so a host that sees
+/// no frames at all for several intervals can conclude the sidecar process is
+/// dead or wedged rather than merely busy. Runs on its own thread with a clone
+/// of the outbound frame channel so beats are independent of the dispatch loop.
+fn spawn_heartbeat_thread(write_tx: TrackedSyncSender<ProtocolFrame>, 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::<ProtocolFrame>(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);

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"

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<Vec<u8>>,
     /// Outbound callback/control response frames. The writer drains this before regular requests.
     control_writer_tx: mpsc::Sender<Vec<u8>>,
+    /// When the reader last received any inbound frame; the silence watchdog reads it.
+    last_inbound_at: parking_lot::Mutex<std::time::Instant>,
 }
 
 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<SidecarTransport>, 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<SidecarTransport>, 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>) -> 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();

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

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 = {

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,

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,

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 {