diff --git a/Cargo.lock b/Cargo.lock
index 60e1e08b32..a2ade3bdd1 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1731,7 +1731,6 @@ name = "epoxy-protocol"
 version = "2.3.7"
 dependencies = [
  "anyhow",
- "rivet-util",
  "rivet-vbare-compiler",
  "serde",
  "serde_bare",
@@ -5340,7 +5339,6 @@ dependencies = [
  "anyhow",
  "gasoline",
  "rivet-runner-protocol",
- "rivet-util",
  "rivet-vbare-compiler",
  "serde",
  "serde_bare",
@@ -5385,7 +5383,6 @@ name = "rivet-depot-protocol"
 version = "2.3.7"
 dependencies = [
  "anyhow",
- "rivet-util",
  "rivet-vbare-compiler",
  "serde",
  "serde_bare",
@@ -5520,7 +5517,6 @@ dependencies = [
  "anyhow",
  "hex",
  "rand 0.8.5",
- "rivet-util",
  "rivet-vbare-compiler",
  "serde",
  "serde_bare",
@@ -5983,7 +5979,6 @@ version = "2.3.7"
 dependencies = [
  "anyhow",
  "base64 0.22.1",
- "rivet-util",
  "rivet-vbare-compiler",
  "serde",
  "serde_bare",
diff --git a/container-runner/Dockerfile.release b/container-runner/Dockerfile.release
index 833ff98325..6b15599d1c 100644
--- a/container-runner/Dockerfile.release
+++ b/container-runner/Dockerfile.release
@@ -9,6 +9,12 @@
 # -> ./dist/rivet-container-runner  (x86_64 ELF)
 
 FROM --platform=linux/amd64 rust:1-bookworm AS builder
+# The build context excludes `.git` (see `.dockerignore`), so pass the commit SHA
+# in to embed it in the binary's startup version log. Defaults to empty, which
+# the build resolves to "unknown".
+#   docker build --build-arg OVERRIDE_GIT_SHA=$(git rev-parse HEAD) ...
+ARG OVERRIDE_GIT_SHA=
+ENV OVERRIDE_GIT_SHA=${OVERRIDE_GIT_SHA}
 WORKDIR /build
 # The crate depends on in-repo workspace crates (rivet-envoy-client), so the whole
 # workspace is the build context. Build from the rivet repo root.
diff --git a/container-runner/README.md b/container-runner/README.md
index 2dfc56f4e7..0a3fda9a8c 100644
--- a/container-runner/README.md
+++ b/container-runner/README.md
@@ -8,7 +8,8 @@ This README covers development, the example projects, and the local test harness
 spawning a child game-server process per actor and proxying Rivet's tunneled
 HTTP/WebSocket traffic to it. Each child gets its own port; the pool's request
 concurrency decides how many actors share a container (one, in the recommended
-game-server setup), and the process exits when the last actor stops. Wrap any dedicated server (Unity, Godot, a plain Node process) in a
+game-server setup). The instance stays warm after its last actor stops; the engine
+reaps it by draining the `/start` connection after the request lifespan. Wrap any dedicated server (Unity, Godot, a plain Node process) in a
 container with this binary as the entrypoint and Rivet Compute can cold-start and route
 to it.
 
diff --git a/container-runner/build.rs b/container-runner/build.rs
new file mode 100644
index 0000000000..dd9e37f919
--- /dev/null
+++ b/container-runner/build.rs
@@ -0,0 +1,34 @@
+//! Captures the git commit SHA at build time and exposes it to the binary as the
+//! `CONTAINER_RUNNER_GIT_SHA` compile-time env var.
+//!
+//! Resolution order, chosen so the build never fails when git is unavailable:
+//!   1. `OVERRIDE_GIT_SHA` env var. The release image build context excludes
+//!      `.git` (see `.dockerignore`), so CI/Docker injects the SHA this way.
+//!   2. A local `git rev-parse HEAD`, for colocated dev builds.
+//!   3. `"unknown"`.
+
+use std::process::Command;
+
+fn main() {
+	println!("cargo:rerun-if-env-changed=OVERRIDE_GIT_SHA");
+
+	let git_sha = std::env::var("OVERRIDE_GIT_SHA")
+		.ok()
+		.filter(|sha| !sha.trim().is_empty())
+		.or_else(git_head_sha)
+		.unwrap_or_else(|| "unknown".to_string());
+
+	println!("cargo:rustc-env=CONTAINER_RUNNER_GIT_SHA={git_sha}");
+}
+
+fn git_head_sha() -> Option<String> {
+	let output = Command::new("git")
+		.args(["rev-parse", "HEAD"])
+		.output()
+		.ok()?;
+	if !output.status.success() {
+		return None;
+	}
+	let sha = String::from_utf8(output.stdout).ok()?.trim().to_string();
+	if sha.is_empty() { None } else { Some(sha) }
+}
diff --git a/container-runner/src/actor.rs b/container-runner/src/actor.rs
index ff9c53f621..78a14ade44 100644
--- a/container-runner/src/actor.rs
+++ b/container-runner/src/actor.rs
@@ -4,9 +4,10 @@
 //! readiness (so the actor is never reported ready before the child listens),
 //! `run` is a watchdog that reports unexpected child exits,
 //! `on_fetch`/`on_websocket` proxy tunneled traffic to the child's port, and
-//! `on_destroy` stops the child, exiting the process once no actors remain.
+//! `on_destroy` stops the child while the instance stays warm for the next
+//! placement.
 
-use std::sync::Arc;
+use std::sync::{Arc, LazyLock};
 
 use anyhow::{Context, Result};
 use async_trait::async_trait;
@@ -16,10 +17,16 @@ use tokio::sync::Mutex as TokioMutex;
 use crate::child::{ChildProcess, SpawnSpec, log_prefix};
 use crate::input::ActorInput;
 use crate::{
-	children, effective_stop_grace, release_child_port, request_exit, reserve_child_port,
+	children, effective_stop_grace, release_child_port, reserve_child_port,
 	runner_config,
 };
 
+/// Live actor contexts on this instance, keyed by actor id. Lets the process
+/// shutdown path report actors as crashed when the platform reclaims the
+/// container out from under them.
+static ACTOR_CTXS: LazyLock<scc::HashMap<String, Ctx<GameServer>>> =
+	LazyLock::new(scc::HashMap::new);
+
 pub struct GameServer {
 	child: TokioMutex<Option<Arc<ChildProcess>>>,
 }
@@ -34,17 +41,19 @@ impl GameServer {
 		// deliberate, then stop. `stop` is idempotent if the process shutdown
 		// sweep already stopped this child.
 		children().remove_async(actor_id).await;
+		ACTOR_CTXS.remove_async(actor_id).await;
 		let child = self.child.lock().await.take();
 		if let Some(child) = child {
 			child.stop(effective_stop_grace()).await;
 			release_child_port(child.child_port).await;
 		}
 
-		// Once the last actor is gone the instance drains rather than
-		// lingering for the next placement.
-		if children().is_empty() {
-			request_exit(actor_id, reason);
-		}
+		// The instance stays alive and warm after its last actor stops, ready to
+		// host the next placement. It is reaped by the platform's own shutdown
+		// signal, not by self-exit. This keeps the serverless container long
+		// lived enough for the log agent to drain its stderr, which a fast
+		// self-exit could otherwise lose.
+		tracing::info!(actor_id = %actor_id, reason, "actor stopped, keeping instance warm");
 	}
 }
 
@@ -76,6 +85,17 @@ impl Actor for GameServer {
 		let actor_id = ctx.actor_id().to_string();
 		let key = actor_key_string(&ctx);
 
+		// Surface the resource monitor's status here, tagged with the actor id, so
+		// it is visible in actor-scoped log views. The monitor's own enable/disable
+		// logs are process-level and have no actor id, so they are filtered out of
+		// those views.
+		tracing::info!(
+			actor_id = %actor_id,
+			resource_monitor_enabled = crate::monitor::enabled(),
+			resource_monitor_source = crate::monitor::sampling_source(),
+			"resource monitor status"
+		);
+
 		// An engine retry for an actor that is already running here must be an
 		// idempotent no-op: rejecting it would make the engine tear down a
 		// healthy actor.
@@ -85,6 +105,7 @@ impl Actor for GameServer {
 					"{} runner: actor already running, ignoring duplicate start",
 					log_prefix(&actor_id, existing.key.as_deref())
 				);
+				register_ctx(&actor_id, &ctx).await;
 				*self.child.lock().await = Some(existing);
 				return Ok(());
 			}
@@ -118,6 +139,23 @@ impl Actor for GameServer {
 			key: key.clone(),
 		};
 
+		// Version line, tagged with the actor id so it is visible in actor-scoped
+		// logs. `git_sha` is omitted entirely when unknown rather than logged as
+		// "unknown".
+		match crate::git_sha() {
+			Some(git_sha) => tracing::info!(
+				actor_id = %actor_id,
+				version = crate::VERSION,
+				git_sha = %git_sha,
+				"container-runner build"
+			),
+			None => tracing::info!(
+				actor_id = %actor_id,
+				version = crate::VERSION,
+				"container-runner build"
+			),
+		}
+
 		tracing::info!(
 			boot_id = crate::boot_id(),
 			actor_id = %actor_id,
@@ -129,12 +167,10 @@ impl Actor for GameServer {
 			Ok(child) => Arc::new(child),
 			Err(err) => {
 				release_child_port(child_port).await;
-				// A failed start on an otherwise idle instance poisons it;
-				// don't let it serve the next placement. With other actors
-				// running, the failure is this actor's alone.
-				if children().is_empty() {
-					request_exit(&actor_id, "child failed to start");
-				}
+				// A failed start is this actor's alone and does not take the
+				// instance down. The container stays warm and ready for the next
+				// placement, and stays alive long enough for the log agent to
+				// drain the failure logs before the platform reaps it.
 				return Err(err);
 			}
 		};
@@ -152,6 +188,10 @@ impl Actor for GameServer {
 			release_child_port(child_port).await;
 			anyhow::bail!("a child for actor {actor_id} is already registered");
 		}
+		// Register only now that startup has succeeded. Registering earlier would
+		// leak an entry for any generation whose start failed, since a failed
+		// start never runs on_destroy/on_sleep to remove it.
+		register_ctx(&actor_id, &ctx).await;
 		*self.child.lock().await = Some(child);
 		Ok(())
 	}
@@ -236,6 +276,38 @@ impl Actor for GameServer {
 	}
 }
 
+/// Register an actor context for crash-on-shutdown reporting. Overwrites any
+/// stale entry left by a prior generation with the same id.
+async fn register_ctx(actor_id: &str, ctx: &Ctx<GameServer>) {
+	ACTOR_CTXS.remove_async(actor_id).await;
+	let _ = ACTOR_CTXS
+		.insert_async(actor_id.to_string(), ctx.clone())
+		.await;
+}
+
+/// Report every live actor on this instance as crashed. Called when the
+/// platform reclaims the container (an unexpected SIGTERM) so the reclaim
+/// surfaces as a crash on the engine instead of a silent reallocation. Runs
+/// while the envoy is still connected so the crash reaches the engine.
+pub async fn crash_all_actors(message: &str) {
+	let mut ctxs = Vec::new();
+	ACTOR_CTXS
+		.retain_async(|_, ctx| {
+			ctxs.push(ctx.clone());
+			false
+		})
+		.await;
+	for ctx in ctxs {
+		if let Err(err) = ctx.stop_with_error(message) {
+			tracing::debug!(
+				actor_id = %ctx.actor_id(),
+				error = ?err,
+				"crash-on-shutdown stop_with_error failed"
+			);
+		}
+	}
+}
+
 fn actor_key_string(ctx: &Ctx<GameServer>) -> Option<String> {
 	let key = ctx.key();
 	if key.is_empty() {
diff --git a/container-runner/src/child.rs b/container-runner/src/child.rs
index 64c4785074..c9d5843ae6 100644
--- a/container-runner/src/child.rs
+++ b/container-runner/src/child.rs
@@ -86,7 +86,7 @@ impl ChildProcess {
 				"child port {child_port} is already in use before spawning `{program}`: a \
                  previous game server is still running in this container. container-runner \
                  hosts one actor per container — configure the serverless runner with \
-                 max_concurrent_actors=1 and Cloud Run request concurrency=1."
+                 max_concurrent_actors=1 and platform request concurrency=1."
 			);
 		}
 
diff --git a/container-runner/src/main.rs b/container-runner/src/main.rs
index 07fad243fe..4594598431 100644
--- a/container-runner/src/main.rs
+++ b/container-runner/src/main.rs
@@ -12,12 +12,14 @@
 //! The runner hosts as many concurrent actors as the engine places on it,
 //! each with its own child process on its own port; the pool's request
 //! concurrency decides how many that is (1 in the recommended game-server
-//! setup). When the last actor stops the process exits so the platform reaps
-//! the instance.
+//! setup). The instance stays warm after its last actor stops and never
+//! self-exits; the engine reaps it by draining the `/start` connection once
+//! the request lifespan elapses, or the platform sends a SIGTERM.
 
 mod actor;
 mod child;
 mod input;
+mod monitor;
 mod proxy;
 
 use std::io::Read;
@@ -35,6 +37,27 @@ use tracing_subscriber::EnvFilter;
 use crate::actor::GameServer;
 use crate::child::ChildProcess;
 
+/// Crate version from Cargo.toml (the workspace version).
+pub(crate) const VERSION: &str = env!("CARGO_PKG_VERSION");
+
+/// Git commit SHA baked in at build time (see `build.rs`). `"unknown"` when the
+/// build had neither `OVERRIDE_GIT_SHA` nor a reachable git repo (the release
+/// image build context excludes `.git`).
+pub(crate) const GIT_SHA: &str = env!("CONTAINER_RUNNER_GIT_SHA");
+
+/// The effective git SHA, or `None` when unknown. Prefers the build-time SHA and
+/// falls back to a runtime `OVERRIDE_GIT_SHA` env var, so a deploy that cannot
+/// inject a build arg can still surface it via an environment variable.
+pub(crate) fn git_sha() -> Option<String> {
+	if GIT_SHA != "unknown" {
+		return Some(GIT_SHA.to_string());
+	}
+	std::env::var("OVERRIDE_GIT_SHA")
+		.ok()
+		.map(|sha| sha.trim().to_string())
+		.filter(|sha| !sha.is_empty())
+}
+
 /// Static runner configuration derived from the CLI/env.
 pub struct RunnerConfig {
 	/// Child command template (program + fixed args) from `-- <command...>`.
@@ -68,15 +91,20 @@ static RESERVED_PORTS: LazyLock<scc::HashSet<u16>> = LazyLock::new(scc::HashSet:
 static EXIT: LazyLock<CancellationToken> = LazyLock::new(CancellationToken::new);
 
 /// Set when the process is shutting down because the PLATFORM sent a signal.
-/// Cloud Run gives a container roughly 10 seconds between SIGTERM and SIGKILL,
-/// so every grace period on this path must fit that budget; engine-initiated
-/// stops keep the full configured grace (their budget is the pool's drain
-/// grace period instead).
+/// The hosting platform gives a container only a bounded window (often ~10
+/// seconds) between SIGTERM and SIGKILL, so every grace period on this path must
+/// fit that budget; engine-initiated stops keep the full configured grace
+/// (their budget is the pool's drain grace period instead).
 static SIGNAL_SHUTDOWN: AtomicBool = AtomicBool::new(false);
 
+/// Set when shutdown was triggered by a platform SIGTERM (an instance reclaim),
+/// as opposed to a local SIGINT (developer Ctrl-C). Distinguishes a reclaim
+/// from a manual stop for downstream shutdown handling.
+static PLATFORM_RECLAIM: AtomicBool = AtomicBool::new(false);
+
 /// How long the platform gives this container between SIGTERM and SIGKILL.
-/// Cloud Run defaults to 10 seconds (configurable up to 60 on the service);
-/// keep this in sync with the platform setting via RIVET_SIGTERM_BUDGET_SECS.
+/// Defaults to 10 seconds; keep this in sync with the platform's actual budget
+/// via RIVET_SIGTERM_BUDGET_SECS.
 /// The signal-path teardown splits the budget: ~60% for the engine drain
 /// (whose per-actor stops SIGTERM children with a grace capped at ~40%), 1s
 /// for the straggler sweep, and the rest as margin.
@@ -110,6 +138,20 @@ pub fn children() -> &'static scc::HashMap<String, Arc<ChildProcess>> {
 	&CHILDREN
 }
 
+/// Actor ids currently hosting a running child on this instance. Used to
+/// attribute the instance-wide resource samples to the running actor(s).
+pub async fn active_actor_ids() -> Vec<String> {
+	let mut ids = Vec::new();
+	// `retain_async` returning `true` keeps every entry: a read-only scan.
+	CHILDREN
+		.retain_async(|actor_id, _| {
+			ids.push(actor_id.clone());
+			true
+		})
+		.await;
+	ids
+}
+
 /// Reserve a local port for a new child. An explicit `input.port` is honored
 /// or refused if another child holds it; otherwise the first free port at or
 /// above the CLI default is picked. The reservation guards the window between
@@ -167,12 +209,12 @@ pub fn effective_stop_grace() -> Duration {
 	}
 }
 
-/// End the process. Called when the LAST actor on this instance is gone (or a
-/// failed start poisoned an otherwise idle instance): the instance drains
-/// instead of lingering for the next placement. The runner is PID 1 in the
+/// End the process. Only the platform shutdown signal drives this now: actors
+/// stopping or failing to start no longer exit the instance, so it stays warm
+/// and reusable and its logs have time to drain. The runner is PID 1 in the
 /// image, so exiting stops the container and the platform reaps the instance.
 pub fn request_exit(actor_id: &str, reason: &str) {
-	tracing::info!(actor_id = %actor_id, reason, "actor finished, exiting container");
+	tracing::info!(actor_id = %actor_id, reason, "shutting down container");
 	EXIT.cancel();
 }
 
@@ -222,8 +264,9 @@ fn base64url_nopad(input: &[u8]) -> String {
     long_about = None,
 )]
 struct Args {
-	/// Serverless HTTP front-door port. Rivet Compute injects RIVET_PORT (plain Cloud Run
-	/// uses PORT); resolved in `main` as --port > RIVET_PORT > PORT > 8080.
+	/// Serverless HTTP front-door port. Rivet Compute injects RIVET_PORT (other
+	/// serverless platforms use the conventional PORT); resolved in `main` as
+	/// --port > RIVET_PORT > PORT > 8080.
 	#[arg(long)]
 	port: Option<u16>,
 
@@ -244,7 +287,7 @@ struct Args {
 	base_path: String,
 
 	/// SIGTERM→SIGKILL grace period (seconds) when stopping the child.
-	#[arg(long, env = "RIVET_STOP_GRACE_SECS", default_value_t = 25)]
+	#[arg(long, env = "RIVET_STOP_GRACE_SECS", default_value_t = 10)]
 	stop_grace_secs: u64,
 
 	/// How long (seconds) to wait for the child's port to open before failing start.
@@ -285,7 +328,7 @@ async fn async_main() -> Result<()> {
 	let boot_id = boot_id();
 	tracing::info!(?args, %boot_id, "starting container-runner");
 
-	// Front-door port: Rivet Compute injects RIVET_PORT; plain Cloud Run uses PORT.
+	// Front-door port: Rivet Compute injects RIVET_PORT; other serverless platforms use the conventional PORT.
 	let port = args
 		.port
 		.or_else(|| env_u16("RIVET_PORT"))
@@ -329,6 +372,7 @@ async fn async_main() -> Result<()> {
 
 	let serve_shutdown = CancellationToken::new();
 	spawn_signal_handler();
+	monitor::spawn_resource_monitor();
 
 	let serve = tokio::spawn(serverless_http::serve(
 		runtime.clone(),
@@ -345,7 +389,11 @@ async fn async_main() -> Result<()> {
 	));
 	tracing::info!(port, "container-runner serverless front door listening");
 
-	// Wait for an exit request, then tear down. Two orders depending on why:
+	// Wait for an exit request, then tear down. Only the signal path is live
+	// today: nothing calls `request_exit` except `spawn_signal_handler`, which
+	// sets `SIGNAL_SHUTDOWN` before cancelling `EXIT`, so the `else` branch is
+	// currently unreachable and kept only as a fallback for a future
+	// actor-driven exit.
 	//
 	// Signal (platform is reclaiming the instance): tell the engine FIRST so
 	// it can start re-placing actors immediately. Its per-actor stops run our
@@ -353,12 +401,22 @@ async fn async_main() -> Result<()> {
 	// The drain is bounded so an unreachable engine cannot eat the whole
 	// platform budget; the sweep then catches any child whose hooks never ran.
 	//
-	// Actor-driven exit (last actor stopped or a failed start poisoned an
-	// idle instance): no platform deadline. Children are already reaped by
-	// the hooks (the sweep is a no-op backstop), and the runtime drains
-	// unbounded so the /start SSE flushes its stopping frame cleanly.
+	// Fallback actor-driven exit (unreachable today): no platform deadline.
+	// Children are already reaped by the hooks (the sweep is a no-op backstop),
+	// and the runtime drains unbounded so the /start SSE flushes cleanly.
 	EXIT.cancelled().await;
 	if SIGNAL_SHUTDOWN.load(Ordering::Acquire) {
+		// A platform SIGTERM reclaims this instance. Report every actor as crashed
+		// before draining so an unexpected SIGTERM (OOM or the ~60 minute request
+		// cap) surfaces as a crash on the engine instead of a silent reallocation.
+		// This runs while the envoy is still connected so the crash reaches the
+		// engine. A local SIGINT (Ctrl-C) drains gracefully without a crash.
+		if PLATFORM_RECLAIM.load(Ordering::Acquire) {
+			crate::actor::crash_all_actors(
+				"runner received unexpected platform SIGTERM, likely OOM or running longer than 60 minutes",
+			)
+			.await;
+		}
 		if tokio::time::timeout(signal_drain_timeout(), runtime.shutdown())
 			.await
 			.is_err()
@@ -413,7 +471,30 @@ fn spawn_signal_handler() {
 		let mut sigterm = signal(SignalKind::terminate()).expect("install SIGTERM handler");
 		let mut sigint = signal(SignalKind::interrupt()).expect("install SIGINT handler");
 		tokio::select! {
-			_ = sigterm.recv() => tracing::info!("received SIGTERM"),
+			_ = sigterm.recv() => {
+				PLATFORM_RECLAIM.store(true, Ordering::Release);
+				// Attribute the reclaim to each running actor so it is visible in
+				// actor-scoped logs, not only the process-level log stream.
+				let mut actor_ids = Vec::new();
+				CHILDREN
+					.retain_async(|actor_id, _| {
+						actor_ids.push(actor_id.clone());
+						true
+					})
+					.await;
+				if actor_ids.is_empty() {
+					tracing::error!(
+						"unexpected platform SIGTERM received, likely hitting OOM or running longer than 60 minutes"
+					);
+				} else {
+					for actor_id in actor_ids {
+						tracing::error!(
+							actor_id = %actor_id,
+							"unexpected platform SIGTERM received, likely hitting OOM or running longer than 60 minutes"
+						);
+					}
+				}
+			}
 			_ = sigint.recv() => tracing::info!("received SIGINT"),
 		}
 		SIGNAL_SHUTDOWN.store(true, Ordering::Release);
diff --git a/container-runner/src/monitor.rs b/container-runner/src/monitor.rs
new file mode 100644
index 0000000000..3f9afea9d9
--- /dev/null
+++ b/container-runner/src/monitor.rs
@@ -0,0 +1,326 @@
+//! Periodic instance resource monitor.
+//!
+//! Opt-in via the [`ENABLE_ENV`] environment variable. When enabled it samples
+//! memory and CPU usage every [`sample_interval`] and logs them, so memory
+//! growth toward the limit (and the OOM that follows) is visible in the logs at
+//! fine granularity. Disabled by default so nothing is logged unless explicitly
+//! turned on.
+//!
+//! Memory and CPU are detected independently, because the gVisor sandbox
+//! exposes cgroup v1 memory but no cgroup v2 or cgroup v1 CPU accounting, so the
+//! two counters legitimately come from different sources.
+//!
+//! Only memory *usage* is reported, not a limit or percentage: under gVisor both
+//! the cgroup `memory.limit_in_bytes` and `/proc/meminfo` report the sandbox size
+//! rather than the container's configured limit, so any percentage would be
+//! misleading.
+//!
+//! Memory sources, in preference order:
+//!   - **cgroup v2** `memory.current` (real Linux). Exact.
+//!   - **cgroup v1** `memory/memory.usage_in_bytes` (gVisor sandbox).
+//!     Container-wide usage (all processes plus page cache).
+//!   - **`/proc/meminfo`** last resort. Under gVisor this reflects the whole
+//!     sandbox, not the container, so it is only an approximation.
+//!
+//! CPU sources, in preference order:
+//!   - **cgroup v2** `cpu.stat`.
+//!   - **`/proc/stat`**, the gVisor sandbox fallback (sandbox-wide, approximate).
+//!
+//! If neither a memory nor a CPU source is readable the monitor logs once and
+//! disables itself.
+
+use std::time::{Duration, Instant};
+
+use tokio::time::{MissedTickBehavior, interval};
+
+/// Environment variable that enables the monitor. Unset or a falsey value means
+/// the monitor does not run and nothing is logged. Truthy values are `1`,
+/// `true`, `yes`, and `on` (case-insensitive).
+const ENABLE_ENV: &str = "RIVET_LOG_RESOURCE_USAGE";
+
+/// Default cadence for sampling and logging resource usage, when
+/// [`INTERVAL_ENV`] is unset.
+const DEFAULT_SAMPLE_INTERVAL: Duration = Duration::from_millis(500);
+
+/// Environment variable overriding the sample/log interval, in milliseconds.
+/// Unset, unparseable, or `0` falls back to [`DEFAULT_SAMPLE_INTERVAL`].
+const INTERVAL_ENV: &str = "RIVET_RESOURCE_USAGE_INTERVAL_MS";
+
+/// cgroup v2 memory usage (real Linux).
+const MEMORY_CURRENT_V2: &str = "/sys/fs/cgroup/memory.current";
+/// cgroup v1 memory usage (gVisor sandbox). Container-wide, includes page
+/// cache. The sibling `memory.limit_in_bytes` is deliberately not read: under
+/// gVisor it reports the sandbox size, not the configured limit.
+const MEMORY_USAGE_V1: &str = "/sys/fs/cgroup/memory/memory.usage_in_bytes";
+const CPU_STAT_V2: &str = "/sys/fs/cgroup/cpu.stat";
+
+const PROC_MEMINFO: &str = "/proc/meminfo";
+const PROC_STAT: &str = "/proc/stat";
+
+/// Kernel clock ticks per second, the unit of `/proc/stat` jiffies. 100 on Linux
+/// and gVisor.
+const USER_HZ: u64 = 100;
+
+/// Where the monitor reads memory counters from.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum MemSource {
+	/// cgroup v2 `memory.current` (real Linux).
+	CgroupV2,
+	/// cgroup v1 `memory.usage_in_bytes` (gVisor sandbox).
+	CgroupV1,
+	/// `/proc/meminfo` (sandbox-wide approximation).
+	Proc,
+}
+
+impl MemSource {
+	fn label(self) -> &'static str {
+		match self {
+			MemSource::CgroupV2 => "cgroup_v2",
+			MemSource::CgroupV1 => "cgroup_v1",
+			MemSource::Proc => "proc",
+		}
+	}
+}
+
+/// Where the monitor reads CPU counters from.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum CpuSource {
+	/// cgroup v2 `cpu.stat` (real Linux).
+	CgroupV2,
+	/// `/proc/stat` (gVisor sandbox, which does not expose cgroup CPU).
+	Proc,
+}
+
+impl CpuSource {
+	fn label(self) -> &'static str {
+		match self {
+			CpuSource::CgroupV2 => "cgroup_v2",
+			CpuSource::Proc => "proc",
+		}
+	}
+}
+
+/// Spawn the background resource monitor if enabled via [`ENABLE_ENV`]. A no-op
+/// when disabled (the default). Fire-and-forget for the process lifetime; it
+/// runs until the process exits.
+pub fn spawn_resource_monitor() {
+	if !monitor_enabled() {
+		return;
+	}
+	let sample_interval = sample_interval();
+	tracing::info!(
+		interval_ms = sample_interval.as_millis() as u64,
+		"resource monitor enabled"
+	);
+	tokio::spawn(run_monitor(sample_interval));
+}
+
+/// The configured sample/log interval: [`INTERVAL_ENV`] in milliseconds when set
+/// to a positive integer, otherwise [`DEFAULT_SAMPLE_INTERVAL`].
+fn sample_interval() -> Duration {
+	match std::env::var(INTERVAL_ENV)
+		.ok()
+		.and_then(|value| value.trim().parse::<u64>().ok())
+	{
+		Some(ms) if ms > 0 => Duration::from_millis(ms),
+		_ => DEFAULT_SAMPLE_INTERVAL,
+	}
+}
+
+fn monitor_enabled() -> bool {
+	match std::env::var(ENABLE_ENV) {
+		Ok(value) => matches!(
+			value.trim().to_ascii_lowercase().as_str(),
+			"1" | "true" | "yes" | "on"
+		),
+		Err(_) => false,
+	}
+}
+
+/// Whether the monitor is turned on via [`ENABLE_ENV`]. Exposed so the actor can
+/// report the monitor's status tagged with its id (process-level monitor logs
+/// are otherwise invisible in actor-scoped log views).
+pub fn enabled() -> bool {
+	monitor_enabled()
+}
+
+/// The memory counter source the monitor would use right now: `"cgroup_v2"`,
+/// `"cgroup_v1"`, `"proc"`, or `"none"`. Exposed for the actor-scoped status log.
+/// Memory is the monitor's headline signal, so this reports the memory source.
+pub fn sampling_source() -> &'static str {
+	match detect_mem_source() {
+		Some(source) => source.label(),
+		None => "none",
+	}
+}
+
+/// Pick the memory source, preferring exact cgroup v2, then cgroup v1 (gVisor
+/// sandbox), then the sandbox-wide `/proc/meminfo` approximation.
+fn detect_mem_source() -> Option<MemSource> {
+	if read_u64(MEMORY_CURRENT_V2).is_some() {
+		Some(MemSource::CgroupV2)
+	} else if read_u64(MEMORY_USAGE_V1).is_some() {
+		Some(MemSource::CgroupV1)
+	} else if read_proc_memory().is_some() {
+		Some(MemSource::Proc)
+	} else {
+		None
+	}
+}
+
+/// Pick the CPU source, preferring cgroup v2 over the `/proc/stat` fallback.
+fn detect_cpu_source() -> Option<CpuSource> {
+	if read_cgroup_cpu_usage_usec().is_some() {
+		Some(CpuSource::CgroupV2)
+	} else if read_proc_cpu_busy_usec().is_some() {
+		Some(CpuSource::Proc)
+	} else {
+		None
+	}
+}
+
+async fn run_monitor(sample_interval: Duration) {
+	let mem_source = detect_mem_source();
+	let cpu_source = detect_cpu_source();
+	if mem_source.is_none() && cpu_source.is_none() {
+		tracing::warn!(
+			cgroup_dir = "/sys/fs/cgroup",
+			proc_meminfo = PROC_MEMINFO,
+			proc_stat = PROC_STAT,
+			"resource monitor disabled: no readable memory or cpu counters"
+		);
+		return;
+	}
+	tracing::info!(
+		mem_source = mem_source.map(MemSource::label).unwrap_or("none"),
+		cpu_source = cpu_source.map(CpuSource::label).unwrap_or("none"),
+		"resource monitor sampling"
+	);
+
+	let mut ticker = interval(sample_interval);
+	// A slow sample must not make the monitor try to catch up with a burst of
+	// back-to-back ticks; just resume on the next boundary.
+	ticker.set_missed_tick_behavior(MissedTickBehavior::Skip);
+	// The first tick fires immediately; consume it so the first logged line
+	// already covers a full interval of CPU time.
+	ticker.tick().await;
+
+	let mut prev_cpu_usec = cpu_source.and_then(cpu_busy_usec);
+	let mut prev_at = Instant::now();
+
+	loop {
+		ticker.tick().await;
+		let now = Instant::now();
+		let elapsed = now.duration_since(prev_at);
+
+		let cpu_now_usec = cpu_source.and_then(cpu_busy_usec);
+		// CPU time used over the interval, divided by wall time, is the number of
+		// vCPU cores consumed (1.0 == one core fully busy).
+		let cpu_cores = match (prev_cpu_usec, cpu_now_usec) {
+			(Some(prev), Some(cur)) if !elapsed.is_zero() => {
+				Some(cur.saturating_sub(prev) as f64 / elapsed.as_micros() as f64)
+			}
+			_ => None,
+		};
+		prev_cpu_usec = cpu_now_usec;
+		prev_at = now;
+
+		// Only log while an actor is running, and attribute the sample to it so it
+		// shows up in that actor's logs. The counters are instance-wide; with more
+		// than one actor on the instance the same sample is logged for each.
+		let actor_ids = crate::active_actor_ids().await;
+		if actor_ids.is_empty() {
+			continue;
+		}
+
+		let mem_used_mib = mem_source.and_then(memory_used_bytes).map(bytes_to_mib);
+
+		for actor_id in actor_ids {
+			tracing::info!(
+				actor_id = %actor_id,
+				mem_source = mem_source.map(MemSource::label).unwrap_or("none"),
+				cpu_source = cpu_source.map(CpuSource::label).unwrap_or("none"),
+				mem_used_mib = ?mem_used_mib,
+				cpu_cores = ?cpu_cores,
+				"instance resource usage"
+			);
+		}
+	}
+}
+
+/// Current memory usage in bytes for the selected source.
+fn memory_used_bytes(source: MemSource) -> Option<u64> {
+	match source {
+		MemSource::CgroupV2 => read_u64(MEMORY_CURRENT_V2),
+		MemSource::CgroupV1 => read_u64(MEMORY_USAGE_V1),
+		MemSource::Proc => read_proc_memory(),
+	}
+}
+
+/// Cumulative busy CPU time in microseconds for the selected source.
+fn cpu_busy_usec(source: CpuSource) -> Option<u64> {
+	match source {
+		CpuSource::CgroupV2 => read_cgroup_cpu_usage_usec(),
+		CpuSource::Proc => read_proc_cpu_busy_usec(),
+	}
+}
+
+/// Read a pseudo-file holding a single unsigned integer. These are in-memory
+/// kernel files, so the synchronous read does not block meaningfully.
+fn read_u64(path: &str) -> Option<u64> {
+	std::fs::read_to_string(path).ok()?.trim().parse().ok()
+}
+
+/// Cumulative CPU time consumed by the cgroup, in microseconds, from the
+/// `usage_usec` line of `cpu.stat`.
+fn read_cgroup_cpu_usage_usec() -> Option<u64> {
+	let stat = std::fs::read_to_string(CPU_STAT_V2).ok()?;
+	stat.lines()
+		.find_map(|line| line.strip_prefix("usage_usec "))
+		.and_then(|value| value.trim().parse().ok())
+}
+
+/// Used memory in bytes from `/proc/meminfo` (`MemTotal - MemAvailable`). Under
+/// gVisor this reflects the whole sandbox, not the container.
+fn read_proc_memory() -> Option<u64> {
+	let total_kb = read_meminfo_kb("MemTotal")?;
+	let available_kb = read_meminfo_kb("MemAvailable")?;
+	Some(total_kb.saturating_sub(available_kb) * 1024)
+}
+
+/// Value in kB of a `/proc/meminfo` key such as `"MemTotal"`.
+fn read_meminfo_kb(key: &str) -> Option<u64> {
+	let content = std::fs::read_to_string(PROC_MEMINFO).ok()?;
+	content.lines().find_map(|line| {
+		let rest = line.strip_prefix(key)?;
+		rest.trim_start_matches(':')
+			.split_whitespace()
+			.next()?
+			.parse()
+			.ok()
+	})
+}
+
+/// Cumulative busy CPU time in microseconds from the aggregate `cpu` line of
+/// `/proc/stat` (`total - idle - iowait`, converted from jiffies).
+fn read_proc_cpu_busy_usec() -> Option<u64> {
+	let content = std::fs::read_to_string(PROC_STAT).ok()?;
+	let line = content.lines().next()?;
+	let mut fields = line.split_whitespace();
+	if fields.next()? != "cpu" {
+		return None;
+	}
+	let values: Vec<u64> = fields.filter_map(|value| value.parse().ok()).collect();
+	if values.len() < 4 {
+		return None;
+	}
+	let total: u64 = values.iter().sum();
+	// Fields are user, nice, system, idle, iowait, ... Treat idle + iowait as idle.
+	let idle = values[3] + values.get(4).copied().unwrap_or(0);
+	let busy = total.saturating_sub(idle);
+	Some(busy * (1_000_000 / USER_HZ))
+}
+
+fn bytes_to_mib(bytes: u64) -> f64 {
+	bytes as f64 / (1024.0 * 1024.0)
+}
diff --git a/engine/sdks/rust/data/Cargo.toml b/engine/sdks/rust/data/Cargo.toml
index f7c8c28c8c..0779ad14e5 100644
--- a/engine/sdks/rust/data/Cargo.toml
+++ b/engine/sdks/rust/data/Cargo.toml
@@ -10,7 +10,6 @@ edition.workspace = true
 anyhow.workspace = true
 gas.workspace = true
 rivet-runner-protocol.workspace = true
-rivet-util.workspace = true
 serde_bare.workspace = true
 serde.workspace = true
 vbare.workspace = true
diff --git a/engine/sdks/rust/data/src/versioned/namespace_runner_config.rs b/engine/sdks/rust/data/src/versioned/namespace_runner_config.rs
index f887175e90..26097a4bd2 100644
--- a/engine/sdks/rust/data/src/versioned/namespace_runner_config.rs
+++ b/engine/sdks/rust/data/src/versioned/namespace_runner_config.rs
@@ -31,22 +31,22 @@ impl OwnedVersionedData for NamespaceRunnerConfig {
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
 			1 => Ok(NamespaceRunnerConfig::V1(
-				rivet_util::serde::bare_from_slice!(payload)?,
+				serde_bare::from_slice(payload)?,
 			)),
 			2 => Ok(NamespaceRunnerConfig::V2(
-				rivet_util::serde::bare_from_slice!(payload)?,
+				serde_bare::from_slice(payload)?,
 			)),
 			3 => Ok(NamespaceRunnerConfig::V3(
-				rivet_util::serde::bare_from_slice!(payload)?,
+				serde_bare::from_slice(payload)?,
 			)),
 			4 => Ok(NamespaceRunnerConfig::V4(
-				rivet_util::serde::bare_from_slice!(payload)?,
+				serde_bare::from_slice(payload)?,
 			)),
 			5 => Ok(NamespaceRunnerConfig::V5(
-				rivet_util::serde::bare_from_slice!(payload)?,
+				serde_bare::from_slice(payload)?,
 			)),
 			6 => Ok(NamespaceRunnerConfig::V6(
-				rivet_util::serde::bare_from_slice!(payload)?,
+				serde_bare::from_slice(payload)?,
 			)),
 			_ => bail!("invalid version: {version}"),
 		}
@@ -55,22 +55,22 @@ impl OwnedVersionedData for NamespaceRunnerConfig {
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
 			NamespaceRunnerConfig::V1(data) => {
-				rivet_util::serde::bare_to_vec!(&data).map_err(Into::into)
+				serde_bare::to_vec(&data).map_err(Into::into)
 			}
 			NamespaceRunnerConfig::V2(data) => {
-				rivet_util::serde::bare_to_vec!(&data).map_err(Into::into)
+				serde_bare::to_vec(&data).map_err(Into::into)
 			}
 			NamespaceRunnerConfig::V3(data) => {
-				rivet_util::serde::bare_to_vec!(&data).map_err(Into::into)
+				serde_bare::to_vec(&data).map_err(Into::into)
 			}
 			NamespaceRunnerConfig::V4(data) => {
-				rivet_util::serde::bare_to_vec!(&data).map_err(Into::into)
+				serde_bare::to_vec(&data).map_err(Into::into)
 			}
 			NamespaceRunnerConfig::V5(data) => {
-				rivet_util::serde::bare_to_vec!(&data).map_err(Into::into)
+				serde_bare::to_vec(&data).map_err(Into::into)
 			}
 			NamespaceRunnerConfig::V6(data) => {
-				rivet_util::serde::bare_to_vec!(&data).map_err(Into::into)
+				serde_bare::to_vec(&data).map_err(Into::into)
 			}
 		}
 	}
diff --git a/engine/sdks/rust/depot-protocol/Cargo.toml b/engine/sdks/rust/depot-protocol/Cargo.toml
index 99d8117349..274836220a 100644
--- a/engine/sdks/rust/depot-protocol/Cargo.toml
+++ b/engine/sdks/rust/depot-protocol/Cargo.toml
@@ -8,7 +8,6 @@ edition.workspace = true
 
 [dependencies]
 anyhow.workspace = true
-rivet-util.workspace = true
 serde_bare.workspace = true
 serde.workspace = true
 vbare.workspace = true
diff --git a/engine/sdks/rust/depot-protocol/src/versioned.rs b/engine/sdks/rust/depot-protocol/src/versioned.rs
index d90a617eec..c8432da5ae 100644
--- a/engine/sdks/rust/depot-protocol/src/versioned.rs
+++ b/engine/sdks/rust/depot-protocol/src/versioned.rs
@@ -22,14 +22,14 @@ impl OwnedVersionedData for DBHead {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
 			_ => bail!("invalid depot db head version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
+			Self::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
 		}
 	}
 }
diff --git a/engine/sdks/rust/envoy-protocol/Cargo.toml b/engine/sdks/rust/envoy-protocol/Cargo.toml
index 907353804d..d0da835e5e 100644
--- a/engine/sdks/rust/envoy-protocol/Cargo.toml
+++ b/engine/sdks/rust/envoy-protocol/Cargo.toml
@@ -12,7 +12,6 @@ description = "Versioned Envoy protocol types for Rivet actor hosts"
 anyhow.workspace = true
 hex.workspace = true
 rand.workspace = true
-rivet-util.workspace = true
 serde_bare.workspace = true
 serde.workspace = true
 utoipa.workspace = true
diff --git a/engine/sdks/rust/envoy-protocol/src/versioned/mod.rs b/engine/sdks/rust/envoy-protocol/src/versioned/mod.rs
index 5357809472..fb5b57ed98 100644
--- a/engine/sdks/rust/envoy-protocol/src/versioned/mod.rs
+++ b/engine/sdks/rust/envoy-protocol/src/versioned/mod.rs
@@ -129,24 +129,24 @@ impl OwnedVersionedData for ToEnvoy {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
-			2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)),
-			3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)),
-			4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)),
-			5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)),
-			6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)),
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(Self::V2(serde_bare::from_slice(payload)?)),
+			3 => Ok(Self::V3(serde_bare::from_slice(payload)?)),
+			4 => Ok(Self::V4(serde_bare::from_slice(payload)?)),
+			5 => Ok(Self::V5(serde_bare::from_slice(payload)?)),
+			6 => Ok(Self::V6(serde_bare::from_slice(payload)?)),
 			_ => bail!("invalid version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
+			Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into),
 		}
 	}
 
@@ -261,24 +261,24 @@ impl OwnedVersionedData for ToRivet {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
-			2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)),
-			3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)),
-			4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)),
-			5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)),
-			6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)),
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(Self::V2(serde_bare::from_slice(payload)?)),
+			3 => Ok(Self::V3(serde_bare::from_slice(payload)?)),
+			4 => Ok(Self::V4(serde_bare::from_slice(payload)?)),
+			5 => Ok(Self::V5(serde_bare::from_slice(payload)?)),
+			6 => Ok(Self::V6(serde_bare::from_slice(payload)?)),
 			_ => bail!("invalid version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
+			Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into),
 		}
 	}
 
@@ -393,24 +393,24 @@ impl OwnedVersionedData for ToEnvoyConn {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
-			2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)),
-			3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)),
-			4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)),
-			5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)),
-			6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)),
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(Self::V2(serde_bare::from_slice(payload)?)),
+			3 => Ok(Self::V3(serde_bare::from_slice(payload)?)),
+			4 => Ok(Self::V4(serde_bare::from_slice(payload)?)),
+			5 => Ok(Self::V5(serde_bare::from_slice(payload)?)),
+			6 => Ok(Self::V6(serde_bare::from_slice(payload)?)),
 			_ => bail!("invalid version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
+			Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into),
 		}
 	}
 
@@ -525,24 +525,24 @@ impl OwnedVersionedData for ToGateway {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
-			2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)),
-			3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)),
-			4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)),
-			5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)),
-			6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)),
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(Self::V2(serde_bare::from_slice(payload)?)),
+			3 => Ok(Self::V3(serde_bare::from_slice(payload)?)),
+			4 => Ok(Self::V4(serde_bare::from_slice(payload)?)),
+			5 => Ok(Self::V5(serde_bare::from_slice(payload)?)),
+			6 => Ok(Self::V6(serde_bare::from_slice(payload)?)),
 			_ => bail!("invalid version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
+			Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into),
 		}
 	}
 
@@ -657,24 +657,24 @@ impl OwnedVersionedData for ToOutbound {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
-			2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)),
-			3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)),
-			4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)),
-			5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)),
-			6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)),
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(Self::V2(serde_bare::from_slice(payload)?)),
+			3 => Ok(Self::V3(serde_bare::from_slice(payload)?)),
+			4 => Ok(Self::V4(serde_bare::from_slice(payload)?)),
+			5 => Ok(Self::V5(serde_bare::from_slice(payload)?)),
+			6 => Ok(Self::V6(serde_bare::from_slice(payload)?)),
 			_ => bail!("invalid version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
+			Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into),
 		}
 	}
 
@@ -789,24 +789,24 @@ impl OwnedVersionedData for ActorCommandKeyData {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(Self::V1(rivet_util::serde::bare_from_slice!(payload)?)),
-			2 => Ok(Self::V2(rivet_util::serde::bare_from_slice!(payload)?)),
-			3 => Ok(Self::V3(rivet_util::serde::bare_from_slice!(payload)?)),
-			4 => Ok(Self::V4(rivet_util::serde::bare_from_slice!(payload)?)),
-			5 => Ok(Self::V5(rivet_util::serde::bare_from_slice!(payload)?)),
-			6 => Ok(Self::V6(rivet_util::serde::bare_from_slice!(payload)?)),
+			1 => Ok(Self::V1(serde_bare::from_slice(payload)?)),
+			2 => Ok(Self::V2(serde_bare::from_slice(payload)?)),
+			3 => Ok(Self::V3(serde_bare::from_slice(payload)?)),
+			4 => Ok(Self::V4(serde_bare::from_slice(payload)?)),
+			5 => Ok(Self::V5(serde_bare::from_slice(payload)?)),
+			6 => Ok(Self::V6(serde_bare::from_slice(payload)?)),
 			_ => bail!("invalid version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Self::V1(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V2(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V3(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V4(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V5(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
-			Self::V6(x) => rivet_util::serde::bare_to_vec!(&x).map_err(Into::into),
+			Self::V1(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V2(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V3(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V4(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V5(x) => serde_bare::to_vec(&x).map_err(Into::into),
+			Self::V6(x) => serde_bare::to_vec(&x).map_err(Into::into),
 		}
 	}
 
@@ -934,7 +934,7 @@ mod tests {
 
 	#[test]
 	fn v1_start_command_deserializes_into_latest_without_sqlite_startup_data() -> Result<()> {
-		let payload = rivet_util::serde::bare_to_vec!(&v1::ToEnvoy::ToEnvoyCommands(vec![
+		let payload = serde_bare::to_vec(&v1::ToEnvoy::ToEnvoyCommands(vec![
 			v1::CommandWrapper {
 				checkpoint: v1::ActorCheckpoint {
 					actor_id: "actor".into(),
@@ -970,7 +970,7 @@ mod tests {
 
 	#[test]
 	fn v2_sqlite_response_does_not_deserialize_to_stateless_protocol() -> Result<()> {
-		let payload = rivet_util::serde::bare_to_vec!(&v2::ToEnvoy::ToEnvoySqliteCommitResponse(
+		let payload = serde_bare::to_vec(&v2::ToEnvoy::ToEnvoySqliteCommitResponse(
 			v2::ToEnvoySqliteCommitResponse {
 				request_id: 1,
 				data: v2::SqliteCommitResponse::SqliteErrorResponse(v2::SqliteErrorResponse {
diff --git a/engine/sdks/rust/epoxy-protocol/Cargo.toml b/engine/sdks/rust/epoxy-protocol/Cargo.toml
index 5903387849..0197f4926d 100644
--- a/engine/sdks/rust/epoxy-protocol/Cargo.toml
+++ b/engine/sdks/rust/epoxy-protocol/Cargo.toml
@@ -8,7 +8,6 @@ edition.workspace = true
 
 [dependencies]
 anyhow.workspace = true
-rivet-util.workspace = true
 serde_bare.workspace = true
 serde.workspace = true
 vbare.workspace = true
diff --git a/engine/sdks/rust/epoxy-protocol/src/versioned.rs b/engine/sdks/rust/epoxy-protocol/src/versioned.rs
index 2dd501bbb8..63d8f736a2 100644
--- a/engine/sdks/rust/epoxy-protocol/src/versioned.rs
+++ b/engine/sdks/rust/epoxy-protocol/src/versioned.rs
@@ -26,10 +26,10 @@ impl OwnedVersionedData for CommittedValue {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			2 => Ok(CommittedValue::V2(rivet_util::serde::bare_from_slice!(
+			2 => Ok(CommittedValue::V2(serde_bare::from_slice(
 				payload
 			)?)),
-			3 => Ok(CommittedValue::V3(rivet_util::serde::bare_from_slice!(
+			3 => Ok(CommittedValue::V3(serde_bare::from_slice(
 				payload
 			)?)),
 			_ => bail!("invalid version: {version}"),
@@ -38,8 +38,8 @@ impl OwnedVersionedData for CommittedValue {
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			CommittedValue::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
-			CommittedValue::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
+			CommittedValue::V2(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			CommittedValue::V3(data) => serde_bare::to_vec(&data).map_err(Into::into),
 		}
 	}
 
@@ -93,10 +93,10 @@ impl OwnedVersionedData for CachedValue {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			2 => Ok(CachedValue::V2(rivet_util::serde::bare_from_slice!(
+			2 => Ok(CachedValue::V2(serde_bare::from_slice(
 				payload
 			)?)),
-			3 => Ok(CachedValue::V3(rivet_util::serde::bare_from_slice!(
+			3 => Ok(CachedValue::V3(serde_bare::from_slice(
 				payload
 			)?)),
 			_ => bail!("invalid version: {version}"),
@@ -105,8 +105,8 @@ impl OwnedVersionedData for CachedValue {
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			CachedValue::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
-			CachedValue::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
+			CachedValue::V2(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			CachedValue::V3(data) => serde_bare::to_vec(&data).map_err(Into::into),
 		}
 	}
 
@@ -161,10 +161,10 @@ impl OwnedVersionedData for AcceptedValue {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			2 => Ok(AcceptedValue::V2(rivet_util::serde::bare_from_slice!(
+			2 => Ok(AcceptedValue::V2(serde_bare::from_slice(
 				payload
 			)?)),
-			3 => Ok(AcceptedValue::V3(rivet_util::serde::bare_from_slice!(
+			3 => Ok(AcceptedValue::V3(serde_bare::from_slice(
 				payload
 			)?)),
 			_ => bail!("invalid version: {version}"),
@@ -173,8 +173,8 @@ impl OwnedVersionedData for AcceptedValue {
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			AcceptedValue::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
-			AcceptedValue::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
+			AcceptedValue::V2(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			AcceptedValue::V3(data) => serde_bare::to_vec(&data).map_err(Into::into),
 		}
 	}
 
@@ -228,16 +228,16 @@ impl OwnedVersionedData for Request {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			2 => Ok(Request::V2(rivet_util::serde::bare_from_slice!(payload)?)),
-			3 => Ok(Request::V3(rivet_util::serde::bare_from_slice!(payload)?)),
+			2 => Ok(Request::V2(serde_bare::from_slice(payload)?)),
+			3 => Ok(Request::V3(serde_bare::from_slice(payload)?)),
 			_ => bail!("invalid version: {version}"),
 		}
 	}
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			Request::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
-			Request::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
+			Request::V2(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			Request::V3(data) => serde_bare::to_vec(&data).map_err(Into::into),
 		}
 	}
 
diff --git a/engine/sdks/rust/ups-protocol/Cargo.toml b/engine/sdks/rust/ups-protocol/Cargo.toml
index 9105a5ea23..f6af0850c4 100644
--- a/engine/sdks/rust/ups-protocol/Cargo.toml
+++ b/engine/sdks/rust/ups-protocol/Cargo.toml
@@ -9,7 +9,6 @@ edition.workspace = true
 [dependencies]
 anyhow.workspace = true
 base64.workspace = true
-rivet-util.workspace = true
 serde_bare.workspace = true
 serde.workspace = true
 vbare.workspace = true
diff --git a/engine/sdks/rust/ups-protocol/src/versioned.rs b/engine/sdks/rust/ups-protocol/src/versioned.rs
index ee756e64bc..a49b575209 100644
--- a/engine/sdks/rust/ups-protocol/src/versioned.rs
+++ b/engine/sdks/rust/ups-protocol/src/versioned.rs
@@ -26,13 +26,13 @@ impl OwnedVersionedData for UpsMessage {
 
 	fn deserialize_version(payload: &[u8], version: u16) -> Result<Self> {
 		match version {
-			1 => Ok(UpsMessage::V1(rivet_util::serde::bare_from_slice!(
+			1 => Ok(UpsMessage::V1(serde_bare::from_slice(
 				payload
 			)?)),
-			2 => Ok(UpsMessage::V2(rivet_util::serde::bare_from_slice!(
+			2 => Ok(UpsMessage::V2(serde_bare::from_slice(
 				payload
 			)?)),
-			3 => Ok(UpsMessage::V3(rivet_util::serde::bare_from_slice!(
+			3 => Ok(UpsMessage::V3(serde_bare::from_slice(
 				payload
 			)?)),
 			_ => bail!("invalid version: {version}"),
@@ -41,9 +41,9 @@ impl OwnedVersionedData for UpsMessage {
 
 	fn serialize_version(self, _version: u16) -> Result<Vec<u8>> {
 		match self {
-			UpsMessage::V1(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
-			UpsMessage::V2(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
-			UpsMessage::V3(data) => rivet_util::serde::bare_to_vec!(&data).map_err(Into::into),
+			UpsMessage::V1(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			UpsMessage::V2(data) => serde_bare::to_vec(&data).map_err(Into::into),
+			UpsMessage::V3(data) => serde_bare::to_vec(&data).map_err(Into::into),
 		}
 	}
 
diff --git a/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs b/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs
index 0c0a53cbdf..c71358969d 100644
--- a/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs
+++ b/rivetkit-rust/packages/rivetkit-core/src/actor/task.rs
@@ -1801,6 +1801,25 @@ impl ActorTask {
 			step = "sync_alarm",
 			"actor shutdown cleanup step completed"
 		);
+		// Destroy cancels the engine-side alarm BEFORE the SQLite teardown.
+		// `cancel_driver_alarm_logged` issues a `set_alarm(None)` that spawns a
+		// last-pushed-alarm SQLite persist; running it here (rather than after
+		// `cleanup_sqlite`) lets `wait_for_pending_alarm_writes` below await that
+		// persist so it cannot race the teardown and fail with
+		// `transaction_closed`. Sleep keeps the persisted engine alarm armed for
+		// the next instance, so it only aborts the local timer, after cleanup.
+		match reason {
+			ShutdownKind::Destroy => {
+				ctx.cancel_driver_alarm_logged();
+				tracing::debug!(
+					actor_id = %actor_id,
+					reason = reason_label,
+					step = "cancel_driver_alarm",
+					"actor shutdown cleanup step completed"
+				);
+			}
+			ShutdownKind::Sleep => {}
+		}
 		ctx.wait_for_pending_alarm_writes().await;
 		tracing::debug!(
 			actor_id = %actor_id,
@@ -1834,15 +1853,7 @@ impl ActorTask {
 					"actor shutdown cleanup step completed"
 				);
 			}
-			ShutdownKind::Destroy => {
-				ctx.cancel_driver_alarm_logged();
-				tracing::debug!(
-					actor_id = %actor_id,
-					reason = reason_label,
-					step = "cancel_driver_alarm",
-					"actor shutdown cleanup step completed"
-				);
-			}
+			ShutdownKind::Destroy => {}
 		}
 		Ok(())
 	}
diff --git a/rivetkit-rust/packages/rivetkit/src/start.rs b/rivetkit-rust/packages/rivetkit/src/start.rs
index faa5368c67..61d8b2789b 100644
--- a/rivetkit-rust/packages/rivetkit/src/start.rs
+++ b/rivetkit-rust/packages/rivetkit/src/start.rs
@@ -7,6 +7,7 @@ use std::time::Duration;
 
 use anyhow::{Context, Result};
 use futures::FutureExt;
+use rivet_error::RivetError;
 use rivetkit_core::actor::ShutdownKind;
 use rivetkit_core::error::{ActorLifecycle, ActorRuntime, action_not_found};
 use rivetkit_core::{ActorEvent, ActorEvents, ActorStart, QueueSendResult, QueueSendStatus, Reply};
@@ -191,23 +192,45 @@ pub async fn run_actor<A: Actor>(start: Start<A>) -> Result<()> {
 		startup_ready,
 	} = start;
 
-	let state = match snapshot.decode()? {
-		Some(state) => state,
-		// Absent input falls back to the input type's default, matching
-		// rivetkit-typescript where createState receives undefined input.
-		None => A::create_state(&ctx, input.decode_or_default()?).await?,
-	};
-	ctx.set_state(state);
-	ctx.clear_state_dirty();
+	// Run the whole startup phase (input decode, state creation, create,
+	// on_create, on_start) as one fallible unit so a failure is forwarded to
+	// the runtime handshake as the real cause instead of being dropped. Without
+	// this, an input decode error would drop `startup_ready`, surfacing only a
+	// generic closed-channel error rather than the actual failure. The failure
+	// itself is logged by rivetkit-core when it drains the run handle.
+	let startup = async {
+		let state = match snapshot.decode()? {
+			Some(state) => state,
+			// Absent input falls back to the input type's default, matching
+			// rivetkit-typescript where createState receives undefined input.
+			None => A::create_state(&ctx, input.decode_or_default()?).await?,
+		};
+		ctx.set_state(state);
+		ctx.clear_state_dirty();
 
-	let actor = Arc::new(A::create(&ctx).await?);
-	if is_new {
-		actor.clone().on_create(ctx.clone()).await?;
-	}
-	actor.clone().on_start(ctx.clone()).await?;
-	if let Some(reply) = startup_ready {
-		let _ = reply.send(Ok(()));
+		let actor = Arc::new(A::create(&ctx).await?);
+		if is_new {
+			actor.clone().on_create(ctx.clone()).await?;
+		}
+		actor.clone().on_start(ctx.clone()).await?;
+		Ok::<_, anyhow::Error>(actor)
 	}
+	.await;
+
+	let actor = match startup {
+		Ok(actor) => {
+			if let Some(reply) = startup_ready {
+				let _ = reply.send(Ok(()));
+			}
+			actor
+		}
+		Err(error) => {
+			if let Some(reply) = startup_ready {
+				let _ = reply.send(Err(anyhow::Error::new(RivetError::extract(&error))));
+			}
+			return Err(error);
+		}
+	};
 
 	let run_cancel = CancellationToken::new();
 	let run_task = spawn_run_task(actor.clone(), ctx.clone(), run_cancel.clone());
@@ -403,7 +426,12 @@ async fn handle_actor_event<A: Actor>(
 				ShutdownKind::Destroy => actor.on_destroy(ctx).await,
 			};
 			reply.send(result);
-			return Ok(true);
+			// Do not end the loop here. Core sends `SerializeState` during the
+			// finalize phase, after this cleanup, to capture any state the hook
+			// wrote (the NAPI runtime keeps its loop alive the same way).
+			// Returning early would drop the event receiver and make that
+			// enqueue fail with `not_ready`. The loop ends when core closes the
+			// channel after `save_final_state`.
 		}
 		ActorEvent::DisconnectConn { conn_id, reply } => {
 			reply.send(ctx.disconnect_conn(&conn_id).await);
@@ -889,6 +917,7 @@ mod tests {
 		);
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -910,6 +939,7 @@ mod tests {
 		);
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -930,6 +960,7 @@ mod tests {
 		assert_eq!(response.status().as_u16(), 404);
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -943,9 +974,39 @@ mod tests {
 		assert_eq!(state.created, 1);
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
+	#[tokio::test]
+	async fn run_actor_invalid_input_fails_to_start() {
+		// Non-empty bytes that are not valid CBOR for the input type must fail
+		// the actor start rather than silently defaulting, and the failure must
+		// be forwarded to the startup handshake instead of dropped.
+		let (_tx, rx) = unbounded_channel();
+		let (mut start, _ctx) =
+			lifecycle_start_with_ctx(Some(vec![0xff, 0xff, 0xff]), None, rx.into());
+		let (ready_tx, ready_rx) = oneshot::channel();
+		start.startup_ready = Some(ready_tx);
+
+		let error = run_actor::<LifecycleActor>(start)
+			.await
+			.expect_err("invalid input should fail actor start");
+
+		assert!(
+			format!("{error:#}").contains("decode actor input from cbor"),
+			"unexpected error: {error:#}"
+		);
+
+		// The startup handshake is signaled with the error rather than dropped,
+		// so the runtime handshake surfaces the real cause. rivetkit-core owns
+		// logging the failure when it drains the run handle.
+		ready_rx
+			.await
+			.expect("startup_ready should be signaled")
+			.expect_err("startup should report failure");
+	}
+
 	#[tokio::test]
 	async fn run_actor_default_websocket_rejects() {
 		let (tx, rx) = unbounded_channel();
@@ -968,6 +1029,7 @@ mod tests {
 		assert!(error.to_string().contains("websockets not supported"));
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -1003,6 +1065,7 @@ mod tests {
 		tx.send(ActorEvent::ConnectionClosed { conn })
 			.expect("send connection closed");
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 
 		let log = &ctx.state().log;
@@ -1032,6 +1095,7 @@ mod tests {
 		assert!(error.to_string().contains("subscribe denied"));
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 
 		let log = &ctx.state().log;
@@ -1066,6 +1130,7 @@ mod tests {
 		assert!(error.to_string().contains("connection rejected"));
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 		assert!(
 			!ctx.state()
@@ -1083,6 +1148,7 @@ mod tests {
 		let actor = tokio::spawn(run_actor::<LifecycleActor>(start));
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 
 		assert!(ctx.state().log.iter().any(|entry| entry == "run_aborted"));
@@ -1096,11 +1162,35 @@ mod tests {
 		let actor = tokio::spawn(run_actor::<LifecycleActor>(start));
 
 		request_destroy(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 
 		assert!(ctx.state().log.iter().any(|entry| entry == "on_destroy"));
 	}
 
+	#[tokio::test]
+	async fn run_actor_serializes_state_after_cleanup() {
+		// Core sends `SerializeState` during finalize, after the cleanup hook,
+		// to capture state the hook wrote. The event loop must stay alive to
+		// handle it instead of ending on the cleanup event (which dropped the
+		// receiver and failed the serialize with `not_ready`).
+		let (tx, rx) = unbounded_channel();
+		let start = lifecycle_start(Some(cbor(&LifecycleInput { count: 5 })), None, rx.into());
+		let actor = tokio::spawn(run_actor::<LifecycleActor>(start));
+
+		request_sleep(&tx).await;
+		let state = decode_actor_state(request_serialize(&tx).await);
+		assert_eq!(state.count, 5);
+		assert!(
+			state.log.iter().any(|entry| entry == "on_sleep"),
+			"serialize after cleanup lost on_sleep state, got: {:?}",
+			state.log
+		);
+
+		drop(tx);
+		actor.await.expect("join run_actor").expect("run actor");
+	}
+
 	#[tokio::test]
 	async fn run_actor_error_requests_errored_stop() {
 		let (tx, rx) = unbounded_channel();
@@ -1248,6 +1338,7 @@ mod tests {
 		);
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -1286,6 +1377,7 @@ mod tests {
 		);
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -1313,6 +1405,7 @@ mod tests {
 		assert_eq!(error.code(), "action_not_found");
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -1342,6 +1435,7 @@ mod tests {
 		);
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -1378,6 +1472,7 @@ mod tests {
 		assert_eq!(error.code(), "not_found");
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
@@ -1419,6 +1514,7 @@ mod tests {
 		);
 
 		request_sleep(&tx).await;
+		drop(tx);
 		actor.await.expect("join run_actor").expect("run actor");
 	}
 
diff --git a/website/src/content/docs/deploy/container-runner.mdx b/website/src/content/docs/deploy/container-runner.mdx
index 239afd17f5..a0494866ce 100644
--- a/website/src/content/docs/deploy/container-runner.mdx
+++ b/website/src/content/docs/deploy/container-runner.mdx
@@ -65,7 +65,7 @@ Create actors against the pool's runner (`default`) and connect clients through
 2. The runner spawns your server as a child process with `PORT` set to the child port, waits for the port to open, and reports the actor as running.
 3. Gateway traffic for the actor arrives over Rivet's tunnel and is proxied to `127.0.0.1:<child port>`. WebSocket clients connect at the bare gateway URL with the `rivet` WebSocket subprotocol. Raw HTTP reaches the child under the `/request/*` prefix on the actor surface (the prefix is stripped before proxying); other paths are reserved for the runtime's own endpoints.
 4. Child stdout and stderr are re-emitted with an `[actorId=... key=...]` prefix so actor logs are attributed in the dashboard.
-5. When an actor stops, the runner sends its child `SIGTERM`, escalates to `SIGKILL` after a grace period, and exits the process once no actors remain.
+5. When an actor stops, the runner sends its child `SIGTERM` and escalates to `SIGKILL` after a grace period. The instance stays warm for the next placement rather than exiting; the engine reaps it by draining the `/start` connection after the request lifespan.
 
 ## Configuration
 
@@ -78,8 +78,9 @@ All flags can also be set through environment variables:
 | `--actor-name` | `RIVET_ACTOR_NAME` | `game` | Actor name this runner serves. |
 | `--runner-version` | `RIVET_RUNNER_VERSION` | `1` | Version reported to the engine, used to drain old runners on deploy. |
 | `--base-path` | `RIVET_SERVERLESS_BASE_PATH` | `/api/rivet` | Base path the engine calls for serverless start. |
-| `--stop-grace-secs` | `RIVET_STOP_GRACE_SECS` | `25` | `SIGTERM` to `SIGKILL` grace period when stopping the child. Capped to a few seconds when the platform itself is reclaiming the instance, so shutdown fits inside the platform's own kill window. |
+| `--stop-grace-secs` | `RIVET_STOP_GRACE_SECS` | `10` | `SIGTERM` to `SIGKILL` grace period when stopping the child. Capped to a fraction of the platform's SIGTERM budget when the platform itself is reclaiming the instance, so shutdown fits inside the platform's own kill window. |
 | `--readiness-timeout-secs` | `RIVET_READINESS_TIMEOUT_SECS` | `30` | How long to wait for the child's port to open before failing the start. |
+| — | `RIVET_SIGTERM_BUDGET_SECS` | `10` | How long the platform gives the container between `SIGTERM` and `SIGKILL`. The signal-path shutdown splits this budget across the engine drain, the child stop grace, and a straggler sweep, so keep it in sync with the platform's actual kill window. |
 
 ### Per-Actor Input
 
@@ -89,11 +90,12 @@ The actor's `input` payload can override the launch spec per actor. All fields a
 {
 	"command": ["./GameServer", "-batchmode"],
 	"args": ["-extra-flag"],
-	"env": { "MATCH_MODE": "ranked" }
+	"env": { "MATCH_MODE": "ranked" },
+	"port": 7770
 }
 ```
 
-`command` replaces the entrypoint command template, `args` are appended to it, and `env` adds environment variables for the child.
+`command` replaces the entrypoint command template, `args` are appended to it, `env` adds environment variables for the child, and `port` overrides the child's listen port (exported to the child as `PORT`), falling back to `--child-port` when omitted.
 
 ## Source and Examples