diff --git a/Cargo.lock b/Cargo.lock
index 03fe5a3029..ff892715b9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -5724,6 +5724,7 @@ dependencies = [
  "serde_json",
  "tokio",
  "tokio-tungstenite",
+ "tokio-util",
  "tracing",
  "tracing-subscriber",
  "urlencoding",
diff --git a/engine/sdks/rust/envoy-client/Cargo.toml b/engine/sdks/rust/envoy-client/Cargo.toml
index 57c28b36a8..f7a534f4f1 100644
--- a/engine/sdks/rust/envoy-client/Cargo.toml
+++ b/engine/sdks/rust/envoy-client/Cargo.toml
@@ -29,6 +29,7 @@ serde.workspace = true
 serde_bare.workspace = true
 serde_json.workspace = true
 tokio-tungstenite = { workspace = true, optional = true }
+tokio-util.workspace = true
 tracing.workspace = true
 urlencoding.workspace = true
 vbare.workspace = true
diff --git a/engine/sdks/rust/envoy-client/src/actor.rs b/engine/sdks/rust/envoy-client/src/actor.rs
index 47dbb6b8c7..097d1ae0ca 100644
--- a/engine/sdks/rust/envoy-client/src/actor.rs
+++ b/engine/sdks/rust/envoy-client/src/actor.rs
@@ -8,6 +8,7 @@ use rivet_envoy_protocol as protocol;
 use tokio::sync::oneshot::error::TryRecvError;
 use tokio::sync::{mpsc, oneshot};
 use tokio::task::JoinSet;
+use tokio_util::sync::CancellationToken;
 use tracing::Instrument;
 
 use crate::connection::ws_send;
@@ -115,6 +116,8 @@ struct ActorContext {
 	ws_entries: BufferMap<WsEntry>,
 	hibernating_requests: Vec<protocol::HibernatingRequest>,
 	active_http_request_count: Arc<AsyncCounter>,
+	/// Cancelled when this generation is declared lost, by the engine or by the envoy itself.
+	lost: CancellationToken,
 }
 
 struct PendingStop {
@@ -123,6 +126,19 @@ struct PendingStop {
 	stop_message: Option<String>,
 }
 
+const LOST_STOP_MESSAGE: &str = "envoy connection lost";
+
+impl PendingStop {
+	/// A stop that escalates to lost must still report `Stopped(Error)`, even if it started as a
+	/// graceful stop. See `begin_stop` for why `Stopped(Ok)` would destroy the actor.
+	fn mark_lost(&mut self) {
+		self.stop_code = protocol::StopCode::Error;
+		if self.stop_message.is_none() {
+			self.stop_message = Some(LOST_STOP_MESSAGE.to_string());
+		}
+	}
+}
+
 enum StopProgress {
 	Stopped,
 	Pending(PendingStop),
@@ -135,8 +151,13 @@ pub fn create_actor(
 	config: protocol::ActorConfig,
 	hibernating_requests: Vec<protocol::HibernatingRequest>,
 	preloaded_kv: Option<protocol::PreloadedKv>,
-) -> (mpsc::UnboundedSender<ToActor>, Arc<AsyncCounter>) {
+) -> (
+	mpsc::UnboundedSender<ToActor>,
+	CancellationToken,
+	Arc<AsyncCounter>,
+) {
 	let (tx, rx) = mpsc::unbounded_channel();
+	let lost = CancellationToken::new();
 	let active_http_request_count = Arc::new(AsyncCounter::new());
 	spawn_detached(actor_inner(
 		shared,
@@ -148,8 +169,9 @@ pub fn create_actor(
 		tx.clone(),
 		rx,
 		active_http_request_count.clone(),
+		lost.clone(),
 	));
-	(tx, active_http_request_count)
+	(tx, lost, active_http_request_count)
 }
 
 #[tracing::instrument(
@@ -171,6 +193,7 @@ async fn actor_inner(
 	tx: mpsc::UnboundedSender<ToActor>,
 	mut rx: mpsc::UnboundedReceiver<ToActor>,
 	active_http_request_count: Arc<AsyncCounter>,
+	lost: CancellationToken,
 ) {
 	let handle = EnvoyHandle {
 		shared: shared.clone(),
@@ -191,6 +214,7 @@ async fn actor_inner(
 		ws_entries: BufferMap::new(),
 		hibernating_requests,
 		active_http_request_count,
+		lost,
 	};
 	let mut http_request_tasks = JoinSet::new();
 	let mut pending_stop: Option<PendingStop> = None;
@@ -200,12 +224,13 @@ async fn actor_inner(
 	let start_result = shared
 		.config
 		.callbacks
-		.on_actor_start(
+		.on_actor_start_with_lost_signal(
 			handle.clone(),
 			actor_id.clone(),
 			generation,
 			config,
 			preloaded_kv,
+			ctx.lost.clone(),
 		)
 		.await;
 
@@ -290,11 +315,20 @@ async fn actor_inner(
 						command_idx,
 						reason,
 					} => {
-						if pending_stop.is_some() {
-							tracing::warn!(
-								command_idx,
-								"ignoring duplicate stop while actor teardown is in progress"
-							);
+						if let Some(pending) = pending_stop.as_mut() {
+							if matches!(reason, protocol::StopActorReason::Lost) {
+								tracing::warn!(
+									command_idx,
+									"lost stop escalates actor teardown already in progress"
+								);
+								ctx.lost.cancel();
+								pending.mark_lost();
+							} else {
+								tracing::warn!(
+									command_idx,
+									"ignoring duplicate stop while actor teardown is in progress"
+								);
+							}
 							continue;
 						}
 						if command_idx <= ctx.command_idx {
@@ -308,20 +342,23 @@ async fn actor_inner(
 						}
 					}
 					ToActor::Lost => {
-						if pending_stop.is_some() {
-							tracing::warn!(
-								"ignoring lost signal while actor teardown is in progress"
-							);
+						ctx.lost.cancel();
+						if let Some(pending) = pending_stop.as_mut() {
+							tracing::warn!("lost signal escalates actor teardown already in progress");
+							pending.mark_lost();
 							continue;
 						}
 
 						ctx.error = Some("actor lost due to timeout".to_string());
 
+						// The engine treats this generation as lost once the envoy has been
+						// disconnected past the lost threshold, so stop it the same way as an
+						// engine-sent Lost.
 						match begin_stop(
 							&mut ctx,
 							&handle,
 							&mut http_request_tasks,
-							protocol::StopActorReason::SleepIntent,
+							protocol::StopActorReason::Lost,
 						)
 						.await
 						{
@@ -447,10 +484,10 @@ async fn begin_stop(
 	// `envoy_lost_threshold`.
 	let (mut stop_code, mut stop_message) = if let Some(err) = ctx.error.clone() {
 		(protocol::StopCode::Error, Some(err))
-	} else if matches!(reason, protocol::StopActorReason::Lost) {
+	} else if matches!(reason, protocol::StopActorReason::Lost) || ctx.lost.is_cancelled() {
 		(
 			protocol::StopCode::Error,
-			Some("envoy connection lost".to_string()),
+			Some(LOST_STOP_MESSAGE.to_string()),
 		)
 	} else {
 		(protocol::StopCode::Ok, None)
@@ -540,9 +577,17 @@ fn send_stopped_event_for_result(
 
 fn send_stopped_event(
 	ctx: &mut ActorContext,
-	stop_code: protocol::StopCode,
-	stop_message: Option<String>,
+	mut stop_code: protocol::StopCode,
+	mut stop_message: Option<String>,
 ) {
+	// The lost token is cancelled synchronously before the Lost command is queued, so a graceful
+	// stop can complete after the generation was lost but before this task consumed the command.
+	// Recheck here so a lost generation never reports Stopped(Ok), which the engine treats as a
+	// destroy.
+	if ctx.lost.is_cancelled() && matches!(stop_code, protocol::StopCode::Ok) {
+		stop_code = protocol::StopCode::Error;
+		stop_message = Some(LOST_STOP_MESSAGE.to_string());
+	}
 	send_event(
 		ctx,
 		protocol::Event::EventActorStateUpdate(protocol::EventActorStateUpdate {
@@ -1717,6 +1762,33 @@ mod tests {
 		.expect("timed out waiting for tunnel message")
 	}
 
+	async fn wait_for_stopped_code(
+		envoy_rx: &mut mpsc::UnboundedReceiver<ToEnvoyMessage>,
+	) -> protocol::StopCode {
+		tokio::time::timeout(Duration::from_secs(2), async {
+			loop {
+				let Some(msg) = envoy_rx.recv().await else {
+					panic!("envoy channel closed before stopped event");
+				};
+				let ToEnvoyMessage::SendEvents { events } = msg else {
+					continue;
+				};
+				for event in events {
+					if let protocol::Event::EventActorStateUpdate(
+						protocol::EventActorStateUpdate {
+							state: protocol::ActorState::ActorStateStopped(stopped),
+						},
+					) = event.inner
+					{
+						return stopped.code;
+					}
+				}
+			}
+		})
+		.await
+		.expect("timed out waiting for stopped event")
+	}
+
 	pub(super) async fn wait_for_stopped_event(
 		envoy_rx: &mut mpsc::UnboundedReceiver<ToEnvoyMessage>,
 	) {
@@ -1823,7 +1895,7 @@ mod tests {
 			release_fetch.clone(),
 		));
 		let (shared, mut envoy_rx) = build_shared_context(callbacks);
-		let (actor_tx, active_http_request_count) = create_actor(
+		let (actor_tx, _lost, active_http_request_count) = create_actor(
 			shared,
 			"actor-1".to_string(),
 			1,
@@ -1864,7 +1936,7 @@ mod tests {
 		let (fetch_dropped_tx, fetch_dropped_rx) = oneshot::channel();
 		let callbacks = Arc::new(TestCallbacks::hanging(fetch_started_tx, fetch_dropped_tx));
 		let (shared, mut envoy_rx) = build_shared_context(callbacks);
-		let (actor_tx, active_http_request_count) = create_actor(
+		let (actor_tx, _lost, active_http_request_count) = create_actor(
 			shared,
 			"actor-2".to_string(),
 			1,
@@ -1909,7 +1981,7 @@ mod tests {
 			stop_handle_tx: Mutex::new(Some(stop_handle_tx)),
 		});
 		let (shared, mut envoy_rx) = build_shared_context(callbacks);
-		let (actor_tx, _active_http_request_count) = create_actor(
+		let (actor_tx, _lost, _active_http_request_count) = create_actor(
 			shared,
 			"actor-3".to_string(),
 			1,
@@ -1942,7 +2014,7 @@ mod tests {
 			stop_handle_tx: Mutex::new(Some(stop_handle_tx)),
 		});
 		let (shared, mut envoy_rx) = build_shared_context(callbacks);
-		let (actor_tx, _active_http_request_count) = create_actor(
+		let (actor_tx, _lost, _active_http_request_count) = create_actor(
 			shared,
 			"actor-4".to_string(),
 			1,
@@ -1979,6 +2051,112 @@ mod tests {
 		assert!(stop_handle.complete(), "stop handle should complete once");
 		assert_alarm_before_stopped_event(&mut envoy_rx, Some(123)).await;
 	}
+	#[tokio::test]
+	async fn lost_signal_cancels_the_generation_token() {
+		let (stop_handle_tx, _stop_handle_rx) = oneshot::channel();
+		let callbacks = Arc::new(DeferredStopCallbacks {
+			stop_handle_tx: Mutex::new(Some(stop_handle_tx)),
+		});
+		let (shared, _envoy_rx) = build_shared_context(callbacks);
+		let (actor_tx, lost, _active_http_request_count) = create_actor(
+			shared,
+			"actor-lost-token".to_string(),
+			1,
+			actor_config(),
+			Vec::new(),
+			None,
+		);
+		assert!(!lost.is_cancelled());
+
+		actor_tx.send(ToActor::Lost).expect("failed to send lost");
+		tokio::time::timeout(Duration::from_secs(2), lost.cancelled())
+			.await
+			.expect("lost signal should cancel the generation token");
+	}
+
+	#[tokio::test]
+	async fn lost_during_graceful_stop_reports_stopped_error() {
+		let (stop_handle_tx, stop_handle_rx) = oneshot::channel();
+		let callbacks = Arc::new(DeferredStopCallbacks {
+			stop_handle_tx: Mutex::new(Some(stop_handle_tx)),
+		});
+		let (shared, mut envoy_rx) = build_shared_context(callbacks);
+		let (actor_tx, lost, _active_http_request_count) = create_actor(
+			shared,
+			"actor-lost-escalation".to_string(),
+			1,
+			actor_config(),
+			Vec::new(),
+			None,
+		);
+
+		actor_tx
+			.send(ToActor::Stop {
+				command_idx: 1,
+				reason: protocol::StopActorReason::SleepIntent,
+			})
+			.expect("failed to send sleep stop");
+		let stop_handle = tokio::time::timeout(Duration::from_secs(2), stop_handle_rx)
+			.await
+			.expect("timed out waiting for stop handle")
+			.expect("stop handle sender dropped");
+
+		// The engine gives up on the generation while its graceful sleep is still running.
+		actor_tx
+			.send(ToActor::Stop {
+				command_idx: 2,
+				reason: protocol::StopActorReason::Lost,
+			})
+			.expect("failed to send lost stop");
+		tokio::time::timeout(Duration::from_secs(2), lost.cancelled())
+			.await
+			.expect("lost stop should cancel the generation token");
+
+		assert!(stop_handle.complete(), "stop handle should complete once");
+		assert_eq!(
+			wait_for_stopped_code(&mut envoy_rx).await,
+			protocol::StopCode::Error,
+			"an escalated stop must not report Stopped(Ok), which the engine treats as a destroy"
+		);
+	}
+
+	#[tokio::test]
+	async fn graceful_stop_completing_before_lost_command_is_consumed_reports_error() {
+		let (stop_handle_tx, stop_handle_rx) = oneshot::channel();
+		let callbacks = Arc::new(DeferredStopCallbacks {
+			stop_handle_tx: Mutex::new(Some(stop_handle_tx)),
+		});
+		let (shared, mut envoy_rx) = build_shared_context(callbacks);
+		let (actor_tx, lost, _active_http_request_count) = create_actor(
+			shared,
+			"actor-lost-race".to_string(),
+			1,
+			actor_config(),
+			Vec::new(),
+			None,
+		);
+
+		actor_tx
+			.send(ToActor::Stop {
+				command_idx: 1,
+				reason: protocol::StopActorReason::SleepIntent,
+			})
+			.expect("failed to send sleep stop");
+		let stop_handle = tokio::time::timeout(Duration::from_secs(2), stop_handle_rx)
+			.await
+			.expect("timed out waiting for stop handle")
+			.expect("stop handle sender dropped");
+
+		// The command path cancels the token before queueing Lost. Complete the graceful stop in
+		// that window, before the actor task sees the Lost command.
+		lost.cancel();
+		assert!(stop_handle.complete(), "stop handle should complete once");
+		assert_eq!(
+			wait_for_stopped_code(&mut envoy_rx).await,
+			protocol::StopCode::Error,
+			"a lost generation must not report Stopped(Ok) even if the Lost command was not consumed"
+		);
+	}
 }
 
 #[cfg(test)]
diff --git a/engine/sdks/rust/envoy-client/src/callbacks.rs b/engine/sdks/rust/envoy-client/src/callbacks.rs
index 0d332ba3d8..d6b65cda97 100644
--- a/engine/sdks/rust/envoy-client/src/callbacks.rs
+++ b/engine/sdks/rust/envoy-client/src/callbacks.rs
@@ -7,6 +7,7 @@ use std::{
 
 use rivet_envoy_protocol as protocol;
 use tokio::sync::oneshot;
+use tokio_util::sync::CancellationToken;
 
 use crate::{
 	handle::EnvoyHandle,
@@ -77,6 +78,21 @@ pub trait EnvoyCallbacks: Send + Sync + 'static {
 		preloaded_kv: Option<protocol::PreloadedKv>,
 	) -> BoxFuture<anyhow::Result<()>>;
 
+	/// Like `on_actor_start`, but also receives the generation's lost signal. envoy-client
+	/// cancels it synchronously when the engine or the envoy itself declares this generation
+	/// lost, before the stop is delivered through `on_actor_stop`.
+	fn on_actor_start_with_lost_signal(
+		&self,
+		handle: EnvoyHandle,
+		actor_id: String,
+		generation: u32,
+		config: protocol::ActorConfig,
+		preloaded_kv: Option<protocol::PreloadedKv>,
+		_lost: CancellationToken,
+	) -> BoxFuture<anyhow::Result<()>> {
+		self.on_actor_start(handle, actor_id, generation, config, preloaded_kv)
+	}
+
 	fn on_actor_stop(
 		&self,
 		_handle: EnvoyHandle,
diff --git a/engine/sdks/rust/envoy-client/src/commands.rs b/engine/sdks/rust/envoy-client/src/commands.rs
index 5d4180e4a8..83043a3891 100644
--- a/engine/sdks/rust/envoy-client/src/commands.rs
+++ b/engine/sdks/rust/envoy-client/src/commands.rs
@@ -51,7 +51,7 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec<protocol::Com
 		match command_wrapper.inner {
 			protocol::Command::CommandStartActor(val) => {
 				let actor_name = val.config.name.clone();
-				let (handle, active_http_request_count) = create_actor(
+				let (handle, lost, active_http_request_count) = create_actor(
 					ctx.shared.clone(),
 					checkpoint.actor_id.clone(),
 					checkpoint.generation,
@@ -60,10 +60,11 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec<protocol::Com
 					val.preloaded_kv,
 				);
 
-				ctx.insert_actor(
+				ctx.insert_actor_with_lost_signal(
 					checkpoint.actor_id.clone(),
 					checkpoint.generation,
 					handle,
+					lost,
 					active_http_request_count,
 					actor_name,
 					checkpoint.index,
@@ -73,6 +74,11 @@ pub async fn handle_commands(ctx: &mut EnvoyContext, commands: Vec<protocol::Com
 				let entry = ctx.get_actor_entry_mut(&checkpoint.actor_id, checkpoint.generation);
 
 				if let Some(entry) = entry {
+					// Signal loss before the stop reaches the actor task, which may be busy
+					// starting or already tearing down.
+					if matches!(val.reason, protocol::StopActorReason::Lost) {
+						entry.lost.cancel();
+					}
 					entry.received_stop = true;
 					entry.last_command_idx = checkpoint.index;
 					let _ = entry.handle.send(crate::actor::ToActor::Stop {
diff --git a/engine/sdks/rust/envoy-client/src/envoy.rs b/engine/sdks/rust/envoy-client/src/envoy.rs
index 9424be9a09..0f1497f033 100644
--- a/engine/sdks/rust/envoy-client/src/envoy.rs
+++ b/engine/sdks/rust/envoy-client/src/envoy.rs
@@ -3,6 +3,7 @@ use std::sync::Arc;
 #[cfg(not(target_arch = "wasm32"))]
 use std::sync::OnceLock;
 use std::sync::atomic::Ordering;
+use tokio_util::sync::CancellationToken;
 
 #[cfg(not(target_arch = "wasm32"))]
 use parking_lot::Mutex;
@@ -93,6 +94,8 @@ pub struct WebSocketRoute {
 
 pub struct ActorEntry {
 	pub handle: mpsc::UnboundedSender<ToActor>,
+	/// Cancelled synchronously when this generation is declared lost.
+	pub lost: CancellationToken,
 	pub active_http_request_count: Arc<AsyncCounter>,
 	pub name: String,
 	pub last_command_idx: i64,
@@ -131,6 +134,8 @@ pub enum ToEnvoyMessage {
 	KvRequest {
 		actor_id: String,
 		data: protocol::KvRequestData,
+		/// Lost signal of the requesting generation. A cancelled request is never sent.
+		lost: Option<CancellationToken>,
 		response_tx: oneshot::Sender<anyhow::Result<protocol::KvResponseData>>,
 	},
 	SqliteRequest {
@@ -234,6 +239,28 @@ impl EnvoyContext {
 		active_http_request_count: Arc<AsyncCounter>,
 		name: String,
 		last_command_idx: i64,
+	) {
+		self.insert_actor_with_lost_signal(
+			actor_id,
+			generation,
+			handle,
+			CancellationToken::new(),
+			active_http_request_count,
+			name,
+			last_command_idx,
+		);
+	}
+
+	/// Registers an actor generation together with the lost signal its actor task observes.
+	pub fn insert_actor_with_lost_signal(
+		&mut self,
+		actor_id: String,
+		generation: u32,
+		handle: mpsc::UnboundedSender<ToActor>,
+		lost: CancellationToken,
+		active_http_request_count: Arc<AsyncCounter>,
+		name: String,
+		last_command_idx: i64,
 	) {
 		let buffered_actor_id = actor_id.clone();
 		let buffered_handle = handle.clone();
@@ -244,6 +271,7 @@ impl EnvoyContext {
 				generation,
 				ActorEntry {
 					handle: handle.clone(),
+					lost,
 					active_http_request_count: active_http_request_count.clone(),
 					name,
 					last_command_idx,
@@ -490,8 +518,8 @@ async fn envoy_loop(
 					ToEnvoyMessage::SendEvents { events } => {
 						handle_send_events(&mut ctx, events).await;
 					}
-					ToEnvoyMessage::KvRequest { actor_id, data, response_tx } => {
-						handle_kv_request(&mut ctx, actor_id, data, response_tx).await;
+					ToEnvoyMessage::KvRequest { actor_id, data, lost, response_tx } => {
+						handle_kv_request(&mut ctx, actor_id, data, lost, response_tx).await;
 					}
 					ToEnvoyMessage::SqliteRequest { request, response_tx } => {
 						handle_sqlite_request(&mut ctx, request, response_tx).await;
@@ -598,6 +626,7 @@ async fn envoy_loop(
 					tracing::warn!("stopping all actors due to envoy lost threshold");
 					for (_actor_id, gens) in &ctx.actors {
 						for (_g, entry) in gens {
+							entry.lost.cancel();
 							if !entry.handle.is_closed() {
 								let _ = entry.handle.send(ToActor::Lost);
 							}
diff --git a/engine/sdks/rust/envoy-client/src/handle.rs b/engine/sdks/rust/envoy-client/src/handle.rs
index 744e56e932..1d94b01f8a 100644
--- a/engine/sdks/rust/envoy-client/src/handle.rs
+++ b/engine/sdks/rust/envoy-client/src/handle.rs
@@ -4,6 +4,7 @@ use std::sync::atomic::Ordering;
 use crate::async_counter::AsyncCounter;
 use rivet_envoy_protocol as protocol;
 use tokio::sync::oneshot;
+use tokio_util::sync::CancellationToken;
 
 use crate::context::SharedContext;
 use crate::envoy::{ActorInfo, ToEnvoyMessage};
@@ -450,12 +451,33 @@ impl EnvoyHandle {
 		&self,
 		actor_id: String,
 		entries: Vec<(Vec<u8>, Vec<u8>)>,
+	) -> anyhow::Result<()> {
+		self.put_kv(actor_id, entries, None).await
+	}
+
+	/// Like [`Self::kv_put`], but the write is dropped before transmission if `lost` fires
+	/// while it is queued.
+	pub async fn kv_put_fenced(
+		&self,
+		actor_id: String,
+		entries: Vec<(Vec<u8>, Vec<u8>)>,
+		lost: CancellationToken,
+	) -> anyhow::Result<()> {
+		self.put_kv(actor_id, entries, Some(lost)).await
+	}
+
+	async fn put_kv(
+		&self,
+		actor_id: String,
+		entries: Vec<(Vec<u8>, Vec<u8>)>,
+		lost: Option<CancellationToken>,
 	) -> anyhow::Result<()> {
 		let (keys, values): (Vec<_>, Vec<_>) = entries.into_iter().unzip();
 		let response = self
-			.send_kv_request(
+			.send_fenced_kv_request(
 				actor_id,
 				protocol::KvRequestData::KvPutRequest(protocol::KvPutRequest { keys, values }),
+				lost,
 			)
 			.await?;
 		match response {
@@ -813,6 +835,15 @@ impl EnvoyHandle {
 		&self,
 		actor_id: String,
 		data: protocol::KvRequestData,
+	) -> anyhow::Result<protocol::KvResponseData> {
+		self.send_fenced_kv_request(actor_id, data, None).await
+	}
+
+	async fn send_fenced_kv_request(
+		&self,
+		actor_id: String,
+		data: protocol::KvRequestData,
+		lost: Option<CancellationToken>,
 	) -> anyhow::Result<protocol::KvResponseData> {
 		let (tx, rx) = tokio::sync::oneshot::channel();
 		crate::envoy::send_to_envoy_tx(
@@ -820,6 +851,7 @@ impl EnvoyHandle {
 			ToEnvoyMessage::KvRequest {
 				actor_id,
 				data,
+				lost,
 				response_tx: tx,
 			},
 		)
diff --git a/engine/sdks/rust/envoy-client/src/kv.rs b/engine/sdks/rust/envoy-client/src/kv.rs
index 24d109d2f8..a5d7509766 100644
--- a/engine/sdks/rust/envoy-client/src/kv.rs
+++ b/engine/sdks/rust/envoy-client/src/kv.rs
@@ -1,5 +1,6 @@
 use rivet_envoy_protocol as protocol;
 use tokio::sync::oneshot;
+use tokio_util::sync::CancellationToken;
 
 use crate::connection::ws_send;
 use crate::envoy::EnvoyContext;
@@ -9,6 +10,8 @@ pub struct KvRequestEntry {
 	pub actor_id: String,
 	pub data: protocol::KvRequestData,
 	pub response_tx: oneshot::Sender<anyhow::Result<protocol::KvResponseData>>,
+	/// Lost signal of the requesting generation. Checked right before transmission.
+	pub lost: Option<CancellationToken>,
 	pub sent: bool,
 	pub timestamp: crate::time::Instant,
 }
@@ -20,6 +23,7 @@ pub async fn handle_kv_request(
 	ctx: &mut EnvoyContext,
 	actor_id: String,
 	data: protocol::KvRequestData,
+	lost: Option<CancellationToken>,
 	response_tx: oneshot::Sender<anyhow::Result<protocol::KvResponseData>>,
 ) {
 	let request_id = ctx.next_kv_request_id;
@@ -29,6 +33,7 @@ pub async fn handle_kv_request(
 		actor_id,
 		data,
 		response_tx,
+		lost,
 		sent: false,
 		timestamp: crate::time::Instant::now(),
 	};
@@ -75,6 +80,21 @@ pub async fn send_single_kv_request(ctx: &mut EnvoyContext, request_id: u32) {
 	if request.sent {
 		return;
 	}
+	// A request queued before its generation was declared lost must not reach storage, even
+	// when it is resent after a reconnect.
+	if request
+		.lost
+		.as_ref()
+		.is_some_and(CancellationToken::is_cancelled)
+	{
+		if let Some(request) = ctx.kv_requests.remove(&request_id) {
+			METRICS.kv_requests_inflight.dec();
+			let _ = request.response_tx.send(Err(anyhow::anyhow!(
+				"actor generation was declared lost; kv request dropped before sending"
+			)));
+		}
+		return;
+	}
 
 	ws_send(
 		&ctx.shared,
diff --git a/engine/sdks/rust/envoy-client/tests/command_dedup.rs b/engine/sdks/rust/envoy-client/tests/command_dedup.rs
index dee69c23d0..050cdc8562 100644
--- a/engine/sdks/rust/envoy-client/tests/command_dedup.rs
+++ b/engine/sdks/rust/envoy-client/tests/command_dedup.rs
@@ -480,3 +480,90 @@ async fn live_actor_is_reacked_on_each_tick() {
 	assert_eq!(second[0].actor_id, "actor-a");
 	assert_eq!(second[0].index, 3);
 }
+
+#[tokio::test]
+async fn lost_stop_command_cancels_the_generation_signal_before_delivery() {
+	let mut ctx = new_envoy_context();
+	let (actor_tx, mut actor_rx) = mpsc::unbounded_channel::<ToActor>();
+	let old_lost = tokio_util::sync::CancellationToken::new();
+	let new_lost = tokio_util::sync::CancellationToken::new();
+	ctx.insert_actor_with_lost_signal(
+		"actor-lost".to_string(),
+		1,
+		actor_tx.clone(),
+		old_lost.clone(),
+		Arc::new(AsyncCounter::new()),
+		"actor-lost".to_string(),
+		-1,
+	);
+	ctx.insert_actor_with_lost_signal(
+		"actor-lost".to_string(),
+		2,
+		actor_tx,
+		new_lost.clone(),
+		Arc::new(AsyncCounter::new()),
+		"actor-lost".to_string(),
+		-1,
+	);
+
+	let mut lost_stop = stop_command("actor-lost", 1, 5);
+	lost_stop.inner = protocol::Command::CommandStopActor(protocol::CommandStopActor {
+		reason: protocol::StopActorReason::Lost,
+	});
+	handle_commands(&mut ctx, vec![lost_stop]).await;
+
+	assert!(
+		old_lost.is_cancelled(),
+		"a lost stop must signal its generation synchronously"
+	);
+	assert!(
+		!new_lost.is_cancelled(),
+		"a lost stop must not signal other generations of the same actor"
+	);
+	assert!(matches!(
+		actor_rx.try_recv(),
+		Ok(ToActor::Stop {
+			reason: protocol::StopActorReason::Lost,
+			..
+		})
+	));
+}
+
+/// A KV write queued before its generation was declared lost must never be transmitted, even
+/// when the queue is flushed later.
+#[tokio::test]
+async fn queued_kv_write_is_dropped_once_its_generation_is_lost() {
+	let mut ctx = new_envoy_context();
+	let lost = tokio_util::sync::CancellationToken::new();
+	let (response_tx, response_rx) = tokio::sync::oneshot::channel();
+	// No connection is open, so the request stays queued.
+	rivet_envoy_client::kv::handle_kv_request(
+		&mut ctx,
+		"actor-lost-kv".to_string(),
+		protocol::KvRequestData::KvPutRequest(protocol::KvPutRequest {
+			keys: vec![b"key".to_vec()],
+			values: vec![b"value".to_vec()],
+		}),
+		Some(lost.clone()),
+		response_tx,
+	)
+	.await;
+	let request_id = *ctx
+		.kv_requests
+		.keys()
+		.next()
+		.expect("request should be queued");
+
+	lost.cancel();
+	rivet_envoy_client::kv::send_single_kv_request(&mut ctx, request_id).await;
+
+	assert!(
+		ctx.kv_requests.is_empty(),
+		"the lost request must be dropped"
+	);
+	let error = response_rx
+		.await
+		.expect("the caller must get a response")
+		.expect_err("the lost request must fail");
+	assert!(format!("{error:#}").contains("declared lost"));
+}
diff --git a/engine/sdks/rust/envoy-client/tests/support/actor_http_stream.rs b/engine/sdks/rust/envoy-client/tests/support/actor_http_stream.rs
index 6368359f98..822b6a9866 100644
--- a/engine/sdks/rust/envoy-client/tests/support/actor_http_stream.rs
+++ b/engine/sdks/rust/envoy-client/tests/support/actor_http_stream.rs
@@ -1011,7 +1011,7 @@ async fn streamed_request_remains_cancellable_after_upload_finishes() {
 	let (fetch_dropped_tx, fetch_dropped_rx) = oneshot::channel();
 	let callbacks = Arc::new(TestCallbacks::hanging(fetch_started_tx, fetch_dropped_tx));
 	let (shared, _envoy_rx) = build_shared_context(callbacks);
-	let (actor_tx, _) = create_actor(
+	let (actor_tx, _, _) = create_actor(
 		shared,
 		"actor-finished-upload".to_string(),
 		1,
@@ -1067,7 +1067,7 @@ async fn streamed_request_returns_credit_only_after_handler_consumption() {
 	let (shared, _envoy_rx) = build_shared_context(callbacks);
 	let (ws_tx, mut ws_rx) = mpsc::unbounded_channel();
 	let session = crate::connection::install_connection(&shared, ws_tx).await;
-	let (actor_tx, _active_http_request_count) = create_actor(
+	let (actor_tx, _lost, _active_http_request_count) = create_actor(
 		shared,
 		"actor-request-window".to_string(),
 		1,
@@ -1139,7 +1139,7 @@ async fn active_http_request_count_spans_streaming_response_drain() {
 	let (shared, _envoy_rx) = build_shared_context(callbacks);
 	let (ws_tx, mut ws_rx) = mpsc::unbounded_channel();
 	let session = crate::connection::install_connection(&shared, ws_tx).await;
-	let (actor_tx, active_http_request_count) = create_actor(
+	let (actor_tx, _lost, active_http_request_count) = create_actor(
 		shared,
 		"actor-stream".to_string(),
 		1,
@@ -1227,7 +1227,7 @@ async fn connection_close_replays_one_indexed_terminal_abort_after_reconnect() {
 	let (shared, mut envoy_rx) = build_shared_context(callbacks);
 	let (ws_tx, mut ws_rx) = mpsc::unbounded_channel();
 	let session_one = crate::connection::install_connection(&shared, ws_tx).await;
-	let (actor_tx, active_http_request_count) = create_actor(
+	let (actor_tx, _lost, active_http_request_count) = create_actor(
 		shared.clone(),
 		"actor-session-flap".to_owned(),
 		1,
@@ -1353,7 +1353,7 @@ async fn failed_response_write_queues_terminal_before_connclose_is_processed() {
 	let (shared, mut envoy_rx) = build_shared_context(callbacks);
 	let (ws_tx, mut ws_rx) = mpsc::unbounded_channel();
 	let session = crate::connection::install_connection(&shared, ws_tx).await;
-	let (actor_tx, active_http_request_count) = create_actor(
+	let (actor_tx, _lost, active_http_request_count) = create_actor(
 		shared.clone(),
 		"actor-write-failure".to_owned(),
 		1,
@@ -1436,7 +1436,7 @@ async fn streamed_response_stalls_at_window_until_gateway_consumes_bytes() {
 	let (shared, _envoy_rx) = build_shared_context(callbacks);
 	let (ws_tx, mut ws_rx) = mpsc::unbounded_channel();
 	let session = crate::connection::install_connection(&shared, ws_tx).await;
-	let (actor_tx, active_http_request_count) = create_actor(
+	let (actor_tx, _lost, active_http_request_count) = create_actor(
 		shared,
 		"actor-response-window".to_string(),
 		1,