diff --git a/doc/concept/standard.md b/doc/concept/standard.md
index 7da4a25b11..37a959a3ac 100644
--- a/doc/concept/standard.md
+++ b/doc/concept/standard.md
@@ -26,6 +26,13 @@ maps everything else to "not supported" or a harmless equivalent. The
 [moq-lite page](/concept/moq-lite#what-moq-lite-leaves-out) lists the
 differences.
 
+On draft 19 and later, a requester can FIN its request stream while a Rust
+publisher keeps its `SUBSCRIBE` or `SUBSCRIBE_NAMESPACE` active; cancellation
+uses `RESET_STREAM` or `STOP_SENDING`. From draft 17, a subscription
+`REQUEST_UPDATE` can change subscriber priority; other changes are refused with
+`NOT_SUPPORTED` and end the subscription with `UPDATE_FAILED`. Drafts 14 to 18
+retain FIN cancellation.
+
 Rust and JavaScript subscribers accept object extension blocks up to 64 KiB.
 This is an implementation limit, not a limit in the IETF draft. A larger
 declared block stops its subgroup stream with `MALFORMED_TRACK` before reading
diff --git a/rs/moq-net/src/ietf/mod.rs b/rs/moq-net/src/ietf/mod.rs
index 944c277893..5968bef179 100644
--- a/rs/moq-net/src/ietf/mod.rs
+++ b/rs/moq-net/src/ietf/mod.rs
@@ -26,6 +26,7 @@ mod publish;
 mod publish_namespace;
 mod publisher;
 mod request;
+mod request_stream;
 mod session;
 pub mod solicit;
 mod subscribe;
diff --git a/rs/moq-net/src/ietf/publish.rs b/rs/moq-net/src/ietf/publish.rs
index 706fa5a16f..379dbb7bd1 100644
--- a/rs/moq-net/src/ietf/publish.rs
+++ b/rs/moq-net/src/ietf/publish.rs
@@ -128,6 +128,8 @@ pub(crate) enum PublishDoneStatus {
 	Unauthorized,
 	/// The track is no longer being published.
 	TrackEnded,
+	/// A requested subscription update could not be applied.
+	UpdateFailed,
 }
 
 impl PublishDoneStatus {
@@ -148,6 +150,7 @@ impl PublishDoneStatus {
 				Self::InternalError => 0x0,
 				Self::Unauthorized => 0x1,
 				Self::TrackEnded => 0x2,
+				Self::UpdateFailed => 0x8,
 			},
 		}
 	}
diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs
index d42749c8ea..90cb597812 100644
--- a/rs/moq-net/src/ietf/publisher.rs
+++ b/rs/moq-net/src/ietf/publisher.rs
@@ -256,13 +256,18 @@ impl<S: crate::transport::poll::Session> Target<S> {
 		}
 	}
 
-	/// Ready when the peer ends this loop by closing the stream it asked on.
+	/// Ready when the peer ends this loop by cancelling the stream it asked on.
 	///
 	/// An unsolicited loop has no stream of its own to watch and parks here: the
 	/// session driver polling it is what drops it when the session ends.
-	fn poll_closed(&mut self, cx: &mut std::task::Context<'_>) -> Poll<Result<(), Error>> {
+	fn poll_closed(
+		&mut self,
+		finished: &mut bool,
+		version: Version,
+		cx: &mut std::task::Context<'_>,
+	) -> Poll<Result<(), Error>> {
 		match self.stream() {
-			Some(stream) => stream.reader.poll_closed(cx),
+			Some(stream) => super::request_stream::poll_cancel(stream, finished, version, cx),
 			None => Poll::Pending,
 		}
 	}
@@ -696,37 +701,22 @@ where
 				})
 				.await?;
 
-			// Run the track, cancelling on reader close (Unsubscribe or stream close).
+			let mut request_finished = false;
+
+			// Serve the track while reading updates; on draft-19+ only abrupt closure cancels.
 			// The fill (when one was requested) runs alongside on its own fetch stream;
 			// its failures reset that stream and never touch the subscription.
 			let mut track_serve =
 				TrackServe::new(self.session.clone(), track, request_id, self.version, range, timescale);
-			let served = {
-				let serve = async {
-					match fill {
-						Some((fill, cache, timescale)) => {
-							let fill = self.run_fill(request_id, priority, fill, cache, timescale);
-							let track = kio::wait(|waiter| track_serve.poll(waiter));
-							let (res, filled) = futures::join!(track, fill);
-							(res, filled)
-						}
-						None => (kio::wait(|waiter| track_serve.poll(waiter)).await, false),
-					}
-				};
-				let mut serve = std::pin::pin!(serve);
-				let mut closed_session = self.session.clone();
-				kio::wait(|waiter| {
-					if let Poll::Ready(served) = waiter.poll_future(serve.as_mut()) {
-						return Poll::Ready(Some(served));
-					}
-					let mut cx = std::task::Context::from_waker(waiter.waker());
-					if stream.reader.poll_closed(&mut cx).is_ready() || closed_session.poll_closed(&mut cx).is_ready() {
-						return Poll::Ready(None);
-					}
-					Poll::Pending
-				})
-				.await
+			let fill = async {
+				match fill {
+					Some((fill, cache, timescale)) => self.run_fill(request_id, priority, fill, cache, timescale).await,
+					None => false,
+				}
 			};
+			let served = self
+				.run_subscription(&mut stream, &mut track_serve, &mut request_finished, fill)
+				.await;
 
 			let completed = served.is_some();
 			let (res, filled) = served.unwrap_or((Ok(()), false));
@@ -746,7 +736,9 @@ where
 						return Poll::Ready(res);
 					}
 					let mut cx = std::task::Context::from_waker(waiter.waker());
-					if stream.reader.poll_closed(&mut cx).is_ready() || closed_session.poll_closed(&mut cx).is_ready() {
+					if super::request_stream::poll_cancel(&mut stream, &mut request_finished, self.version, &mut cx)
+						.is_ready() || closed_session.poll_closed(&mut cx).is_ready()
+					{
 						return Poll::Ready(Err(Error::Cancel));
 					}
 					Poll::Pending
@@ -765,6 +757,7 @@ where
 			// Send PublishDone
 			let (status, reason) = match &res {
 				Ok(()) => (ietf::PublishDoneStatus::TrackEnded, "track ended"),
+				Err(Error::Unsupported) => (ietf::PublishDoneStatus::UpdateFailed, "update failed"),
 				Err(_) => (ietf::PublishDoneStatus::InternalError, "internal error"),
 			};
 			let _ = stream.writer.encode(&ietf::PublishDone::ID).await;
@@ -788,6 +781,107 @@ where
 		}
 	}
 
+	/// Serve a subscription until the track and any fill finish, applying the requester's
+	/// updates. `None` when the requester or session cancelled it.
+	///
+	/// Drafts 14-16 deliver follow-ups on a virtual stream, where anything still ends the
+	/// subscription. From draft 17 a REQUEST_UPDATE is read and answered: a priority
+	/// change is applied, and anything else is refused and ends the subscription with
+	/// UPDATE_FAILED rather than being dropped silently.
+	async fn run_subscription(
+		&self,
+		stream: &mut Stream<S, Version>,
+		serve: &mut TrackServe<S>,
+		finished: &mut bool,
+		fill: impl std::future::Future<Output = bool>,
+	) -> Option<(Result<(), Error>, bool)> {
+		let legacy = matches!(self.version, Version::Draft14 | Version::Draft15 | Version::Draft16);
+		let mut session = self.session.clone();
+		let mut fill = std::pin::pin!(fill);
+		let mut filled = None;
+		let mut served = None;
+		loop {
+			let event = kio::wait(|waiter| {
+				let mut cx = std::task::Context::from_waker(waiter.waker());
+				if session.poll_closed(&mut cx).is_ready() {
+					return Poll::Ready(Err(Error::Cancel));
+				}
+				// Virtual writers on drafts 14-16 are already closed while the request lives.
+				if !legacy && stream.writer.poll_closed(&mut cx).is_ready() {
+					return Poll::Ready(Err(Error::Cancel));
+				}
+				if !*finished {
+					if legacy {
+						if stream.reader.poll_closed(&mut cx).is_ready() {
+							return Poll::Ready(Err(Error::Cancel));
+						}
+					} else {
+						match stream
+							.reader
+							.poll_decode_maybe::<super::request_stream::Update>(&mut cx)
+						{
+							Poll::Ready(Ok(Some(update))) => return Poll::Ready(Ok(Some(update))),
+							Poll::Ready(Ok(None)) if !super::request_stream::fin_cancels(self.version) => {
+								*finished = true
+							}
+							Poll::Ready(Ok(None)) => return Poll::Ready(Err(Error::Cancel)),
+							Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+							Poll::Pending => {}
+						}
+					}
+				}
+				if filled.is_none()
+					&& let Poll::Ready(done) = waiter.poll_future(fill.as_mut())
+				{
+					filled = Some(done);
+				}
+				if served.is_none()
+					&& let Poll::Ready(done) = serve.poll(waiter)
+				{
+					served = Some(done);
+				}
+				match (&served, filled) {
+					(Some(Ok(())), Some(_)) => Poll::Ready(Ok(None)),
+					(Some(Err(err)), Some(_)) => Poll::Ready(Err(err.clone())),
+					_ => Poll::Pending,
+				}
+			})
+			.await;
+			let filled = filled.unwrap_or(false);
+			let update = match event {
+				Ok(None) => return Some((Ok(()), filled)),
+				Err(Error::Cancel) => return None,
+				Err(err) => return Some((Err(err), filled)),
+				Ok(Some(update)) => update,
+			};
+			if update.unsupported {
+				let result = self
+					.write_subscribe_error(
+						&mut stream.writer,
+						serve.request_id,
+						&Error::Unsupported,
+						"REQUEST_UPDATE parameters not supported",
+					)
+					.await;
+				return Some((result.and(Err(Error::Unsupported)), filled));
+			}
+			if let Some(priority) = update.priority {
+				let mut subscription = serve.track.subscription();
+				subscription.priority = super::priority::from_wire(priority);
+				if let Err(err) = serve.track.update(subscription) {
+					return Some((Err(err), filled));
+				}
+			}
+			let acked = async {
+				stream.writer.encode(&ietf::RequestOk::ID).await?;
+				stream.writer.encode(&ietf::RequestOk { request_id: None }).await
+			};
+			if let Err(err) = acked.await {
+				return Some((Err(err), filled));
+			}
+		}
+	}
+
 	/// Reject a SUBSCRIBE, ending the request stream.
 	///
 	/// Takes the whole stream because delivering the error is the other half of the job:
@@ -2080,9 +2174,10 @@ where
 		let mut retry = crate::runtime::Deadline::new(&self.runtime);
 		let mut retry_at: Option<crate::runtime::Instant> = None;
 		let mut retry_delay = RETRY_BASE;
+		let mut finished = false;
 
-		// Stream updates (origin route (un)announces), bailing if the peer closes
-		// its side first.
+		// Stream updates (origin route (un)announces), bailing if the peer cancels
+		// its request first.
 		let res = loop {
 			match ns.watched.values().any(|watch| watch.deferred) {
 				// Arm on the edge, so a turn that changes nothing else doesn't push the
@@ -2099,7 +2194,7 @@ where
 				let Namespaces { target, .. } = &mut ns;
 				kio::wait(|waiter| {
 					let mut cx = std::task::Context::from_waker(waiter.waker());
-					if let Poll::Ready(res) = target.poll_closed(&mut cx) {
+					if let Poll::Ready(res) = target.poll_closed(&mut finished, self.version, &mut cx) {
 						return Poll::Ready(NamespaceEvent::Closed(res));
 					}
 					if let Poll::Ready(update) = announced.poll_next(waiter) {
@@ -4394,6 +4489,138 @@ mod serve_tests {
 			"the cap excludes cc"
 		);
 	}
+
+	const ALL_VERSIONS: [Version; 9] = [
+		Version::Draft14,
+		Version::Draft15,
+		Version::Draft16,
+		Version::Draft17,
+		Version::Draft18,
+		Version::Draft19,
+		Version::Draft20,
+		Version::Draft21,
+		Version::Draft22,
+	];
+
+	/// Drafts 14-18 cancel a request on its requester's FIN; later drafts only end updates.
+	fn fin_cancels(version: Version) -> bool {
+		matches!(
+			version,
+			Version::Draft14 | Version::Draft15 | Version::Draft16 | Version::Draft17 | Version::Draft18
+		)
+	}
+
+	/// Poll `run` while letting spawned tasks progress; whether it finished.
+	async fn drive<F: std::future::Future>(mut run: std::pin::Pin<&mut F>) -> bool {
+		for _ in 0..50 {
+			if futures::poll!(run.as_mut()).is_ready() {
+				return true;
+			}
+			tokio::task::yield_now().await;
+		}
+		false
+	}
+
+	/// Start a live subscription on a request stream opened from `requester`, whose script
+	/// is what the requester sends after SUBSCRIBE.
+	async fn start_subscription(
+		h: &Serve,
+		mut requester: ScriptedSession,
+	) -> impl std::future::Future<Output = Result<(), Error>> + use<> {
+		settle().await;
+		let stream = Stream::open(&mut requester, h.publisher.version).await.unwrap();
+		h.publisher
+			.clone()
+			.run_subscribe_stream(stream, subscribe(Filter::NextObject, None))
+	}
+
+	/// Draft-21 section 6.4.2.2 lets a requester FIN right after SUBSCRIBE. From draft 19
+	/// that only ends its updates, so media keeps flowing; earlier drafts cancel.
+	#[tokio::test(start_paused = true)]
+	async fn requester_fin_is_version_gated_for_subscriptions() {
+		for version in ALL_VERSIONS {
+			let h = serve(version);
+			let run = start_subscription(&h, ScriptedSession::per_stream_eof(vec![vec![]])).await;
+			let mut run = std::pin::pin!(run);
+			let ended = drive(run.as_mut()).await;
+			assert_eq!(ended, fin_cancels(version), "{version}");
+			if !ended {
+				let mut group = h.track.create_group(group::Info { sequence: 0 }).unwrap();
+				group.write_frame(timestamp(), b"after FIN".as_slice()).unwrap();
+				group.finish().unwrap();
+				assert!(!drive(run.as_mut()).await, "{version}");
+				assert_eq!(occurrences(&h.log, b"after FIN"), 1, "{version}");
+			}
+		}
+	}
+
+	/// A reset request stream still cancels the subscription on every draft.
+	#[tokio::test(start_paused = true)]
+	async fn requester_reset_cancels_subscriptions() {
+		for version in ALL_VERSIONS {
+			let h = serve(version);
+			let run = start_subscription(&h, ScriptedSession::per_stream_reset(vec![vec![]])).await;
+			assert!(drive(std::pin::pin!(run)).await, "{version}");
+		}
+	}
+
+	/// A REQUEST_UPDATE carrying a priority is applied and acknowledged, and the
+	/// subscription keeps going. The second, empty update changes nothing.
+	#[tokio::test(start_paused = true)]
+	async fn subscription_update_applies_priority_without_ending_the_request() {
+		for version in [Version::Draft19, Version::Draft22] {
+			let h = serve(version);
+			let requester = ScriptedSession::per_stream(vec![vec![0x02, 0, 4, 2, 1, 0x20, 10, 0x02, 0, 2, 4, 0]]);
+			let run = start_subscription(&h, requester.clone()).await;
+			let mut run = std::pin::pin!(run);
+			assert!(!drive(run.as_mut()).await, "{version}");
+			assert_eq!(h.track.subscription().unwrap().priority, 245, "{version}");
+			// REQUEST_OK with no parameters, once per update.
+			assert_eq!(occurrences(&requester.log, &[0x07, 0, 1, 0]), 2, "{version}");
+		}
+	}
+
+	/// An update we cannot apply is refused NOT_SUPPORTED and ends the subscription with
+	/// UPDATE_FAILED, rather than silently dropping it.
+	#[tokio::test(start_paused = true)]
+	async fn unsupported_subscription_update_fails_the_subscription() {
+		for version in [Version::Draft17, Version::Draft19, Version::Draft22] {
+			let h = serve(version);
+			// FORWARD=0 asks to pause delivery, which this publisher does not support.
+			let script = match version {
+				// Draft 17 carries a Required Request ID Delta after the Request ID.
+				Version::Draft17 => vec![0x02, 0, 5, 2, 0, 1, 0x10, 0],
+				_ => vec![0x02, 0, 4, 2, 1, 0x10, 0],
+			};
+			let requester = ScriptedSession::per_stream(vec![script]);
+			let run = start_subscription(&h, requester.clone()).await;
+			run.await.unwrap_err();
+			assert_eq!(
+				occurrences(&requester.log, b"REQUEST_UPDATE parameters not supported"),
+				1,
+				"{version}"
+			);
+			assert_eq!(occurrences(&requester.log, b"update failed"), 1, "{version}");
+		}
+	}
+
+	/// SUBSCRIBE_NAMESPACE follows the same gate as SUBSCRIBE.
+	#[tokio::test(start_paused = true)]
+	async fn namespace_requester_fin_is_version_gated() {
+		for version in ALL_VERSIONS {
+			let h = serve(version);
+			let mut requester = ScriptedSession::per_stream_eof(vec![vec![]]);
+			let stream = Stream::open(&mut requester, version).await.unwrap();
+			let msg = ietf::SubscribeNamespace {
+				request_id: RequestId(0),
+				namespace: crate::Path::new(""),
+				hidden: false,
+			};
+			let mut run = std::pin::pin!(h.publisher.clone().run_subscribe_namespace_stream(stream, msg));
+			let ended = drive(run.as_mut()).await;
+			assert_eq!(ended, fin_cancels(version), "{version}");
+		}
+	}
 }
 
 #[cfg(test)]
diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs
new file mode 100644
index 0000000000..bf98967a39
--- /dev/null
+++ b/rs/moq-net/src/ietf/request_stream.rs
@@ -0,0 +1,105 @@
+use std::task::{Context, Poll};
+
+use super::{Message, Version};
+use crate::{
+	Error,
+	coding::{Decode, DecodeError, Stream},
+};
+
+/// Whether the requester's FIN cancels its request; later drafts only end its updates.
+pub(super) fn fin_cancels(version: Version) -> bool {
+	matches!(
+		version,
+		Version::Draft14 | Version::Draft15 | Version::Draft16 | Version::Draft17 | Version::Draft18
+	)
+}
+
+/// Watch abrupt cancellation after the requester has finished sending messages.
+pub(super) fn poll_cancel<S: crate::transport::poll::Session>(
+	stream: &mut Stream<S, Version>,
+	finished: &mut bool,
+	version: Version,
+	cx: &mut Context<'_>,
+) -> Poll<Result<(), Error>> {
+	if !*finished {
+		match stream.reader.poll_closed(cx) {
+			Poll::Ready(Ok(())) if !fin_cancels(version) => *finished = true,
+			Poll::Ready(result) => return Poll::Ready(result),
+			Poll::Pending => {}
+		}
+	}
+	// Virtual writers on drafts 14-16 are already closed while their control request lives.
+	if !matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) {
+		return stream.writer.poll_closed(cx);
+	}
+	Poll::Pending
+}
+
+/// A framed draft-17+ REQUEST_UPDATE on a subscription, preserving omitted preferences.
+#[derive(Debug)]
+pub(super) struct Update {
+	pub(super) priority: Option<u8>,
+	/// It asks for something other than a priority change, which we cannot apply.
+	pub(super) unsupported: bool,
+}
+
+impl Decode<Version> for Update {
+	fn decode<R: bytes::Buf>(r: &mut R, version: Version) -> Result<Self, DecodeError> {
+		if u64::decode(r, version)? != super::SubscribeUpdate::ID {
+			return Err(DecodeError::InvalidValue);
+		}
+		let size = u16::decode(r, version)? as usize;
+		if r.remaining() < size {
+			return Err(DecodeError::Short);
+		}
+		let mut data = r.copy_to_bytes(size);
+		// The complete frame is present; a short field inside it is malformed.
+		Self::decode_body(&mut data, version).map_err(DecodeError::complete)
+	}
+}
+
+impl Update {
+	fn decode_body(data: &mut bytes::Bytes, version: Version) -> Result<Self, DecodeError> {
+		use super::{Fill, Filter, Opaque, RequestId};
+		use bytes::Buf;
+
+		RequestId::decode(data, version)?;
+		if version == Version::Draft17 {
+			// Required Request ID Delta, removed in draft 18.
+			u64::decode(data, version)?;
+		}
+		decode_params!(data, version,
+			0x02 => object_timeout: Option<u64>,
+			0x03 => token: Vec<Opaque>,
+			0x06 => subgroup_timeout: Option<u64>,
+			0x10 => forward: Option<bool>,
+			0x20 => priority: Option<u8>,
+			0x21 => filter: Option<Filter>,
+			0x23 => fill: Option<Fill>,
+			0x25 => subgroup_filter: Vec<Opaque>,
+			0x26 => object_filter: Vec<Opaque>,
+			0x27 => priority_filter: Vec<Opaque>,
+			0x28 => property_filter: Vec<Opaque>,
+			0x29 => track_filter: Vec<Opaque>,
+			0x32 => new_group: Option<u64>,
+		);
+		if data.has_remaining() {
+			return Err(DecodeError::Long);
+		}
+		Ok(Self {
+			priority,
+			unsupported: forward == Some(false)
+				|| filter.is_some()
+				|| fill.is_some()
+				|| object_timeout.is_some()
+				|| subgroup_timeout.is_some()
+				|| !token.is_empty()
+				|| !subgroup_filter.is_empty()
+				|| !object_filter.is_empty()
+				|| !priority_filter.is_empty()
+				|| !property_filter.is_empty()
+				|| !track_filter.is_empty()
+				|| new_group.is_some(),
+		})
+	}
+}
diff --git a/rs/moq-tokio/src/transport.rs b/rs/moq-tokio/src/transport.rs
index 3912a2402b..31d7b34d6b 100644
--- a/rs/moq-tokio/src/transport.rs
+++ b/rs/moq-tokio/src/transport.rs
@@ -55,16 +55,8 @@ const CLOSE_LINGER: std::time::Duration = std::time::Duration::from_secs(10);
 ///   watch parks until the caller's reads drain the backlog, deliberately
 ///   trading watch liveness on an undrained flooding stream for a memory
 ///   bound (every driver in this crate drains its reads alongside the watch).
-///   A send stream only
-///   starts the real watch once `finish` or `reset` makes it terminal; before
-///   that a closure surfaces as an error on the next write instead of waking an
-///   idle watch. The send-side gap this leaves: a peer that resets a stream
-///   sitting idle (no pending write, not finished) does not wake a parked
-///   `poll_closed`, so a driver waiting on "next frame or peer close" learns of
-///   the closure only when the next write fails. The drivers' other arms keep
-///   the session making progress; the stream itself lingers until then. A
-///   native poll implementation observes closure without owning the stream,
-///   which is why a backend that has one is not wrapped.
+///   Send-side watches are interruptible, so an idle watch can observe peer
+///   cancellation while later writes reclaim the stream.
 pub struct Session<S: web_transport_trait::Session> {
 	session: S,
 	accept_uni: OpSlot<Result<S::RecvStream, S::Error>>,
@@ -231,9 +223,6 @@ pub struct SendStream<S: web_transport_trait::SendStream + 'static> {
 	priority: Option<u8>,
 	finish: bool,
 	reset: Option<u32>,
-	/// Whether `finish` or `reset` has been observed, so no further writes can
-	/// come and the closed() watch may safely take ownership of the stream.
-	terminal: bool,
 	/// Bytes already transmitted but not yet reported to the caller: a write the
 	/// stored future finished (possibly inside
 	/// [`poll_closed`](wt_poll::SendStream::poll_closed)) while the caller held a
@@ -273,7 +262,6 @@ impl<S: web_transport_trait::SendStream + 'static> SendStream<S> {
 			priority: None,
 			finish: false,
 			reset: None,
-			terminal: false,
 			completed: Bytes::new(),
 			interrupt: None,
 		}
@@ -374,17 +362,22 @@ impl<S: web_transport_trait::SendStream + 'static> wt_poll::SendStream for SendS
 						}
 					}
 				},
-				SendState::Closing(mut fut) => match fut.as_mut().poll(cx) {
-					Poll::Pending => {
-						*self.state.get_mut().unwrap() = Some(SendState::Closing(fut));
-						return Poll::Pending;
+				SendState::Closing(mut fut) => {
+					if let Some(tx) = self.interrupt.take() {
+						let _ = tx.send(());
 					}
-					Poll::Ready((mut stream, _res)) => {
-						self.interrupt = None;
-						self.settle(&mut stream);
-						*self.state.get_mut().unwrap() = Some(SendState::Idle(stream));
+					match fut.as_mut().poll(cx) {
+						Poll::Pending => {
+							*self.state.get_mut().unwrap() = Some(SendState::Closing(fut));
+							return Poll::Pending;
+						}
+						Poll::Ready((mut stream, _res)) => {
+							self.interrupt = None;
+							self.settle(&mut stream);
+							*self.state.get_mut().unwrap() = Some(SendState::Idle(stream));
+						}
 					}
-				},
+				}
 			}
 		}
 	}
@@ -405,18 +398,21 @@ impl<S: web_transport_trait::SendStream + 'static> wt_poll::SendStream for SendS
 	}
 
 	fn finish(&mut self) -> Result<(), Self::Error> {
-		self.terminal = true;
 		match self.state.get_mut().unwrap().as_mut() {
 			Some(SendState::Idle(stream)) => stream.finish(),
 			_ => {
 				self.finish = true;
+				if matches!(self.state.get_mut().unwrap(), Some(SendState::Closing(_)))
+					&& let Some(tx) = self.interrupt.take()
+				{
+					let _ = tx.send(());
+				}
 				Ok(())
 			}
 		}
 	}
 
 	fn reset(&mut self, code: u32) {
-		self.terminal = true;
 		match self.state.get_mut().unwrap().as_mut() {
 			Some(SendState::Idle(stream)) => stream.reset(code),
 			_ => {
@@ -435,14 +431,6 @@ impl<S: web_transport_trait::SendStream + 'static> wt_poll::SendStream for SendS
 			match self.state.get_mut().unwrap().take().expect("in-flight") {
 				SendState::Idle(mut stream) => {
 					self.settle(&mut stream);
-					// Only a finished (or reset) stream starts the real closed()
-					// watch: that future owns the stream, and a stream the caller
-					// may still write to must stay reclaimable. Before that,
-					// closure surfaces as an error on the next operation instead.
-					if !self.terminal {
-						*self.state.get_mut().unwrap() = Some(SendState::Idle(stream));
-						return Poll::Pending;
-					}
 					// Interruptible like a write: a late reset (the group machines
 					// cancel a finished stream this way) must not wait for a FIN
 					// acknowledgement that may never come.
@@ -839,6 +827,7 @@ mod tests {
 		resets: Arc<Mutex<Vec<u32>>>,
 		/// The peer never acknowledges the FIN: closed() stays pending forever.
 		never_ack: Arc<AtomicBool>,
+		stopped: Arc<AtomicBool>,
 	}
 
 	impl web_transport_trait::SendStream for FakeSend {
@@ -871,10 +860,16 @@ mod tests {
 		}
 
 		async fn closed(&mut self) -> Result<(), Self::Error> {
-			match self.finished.load(Ordering::SeqCst) && !self.never_ack.load(Ordering::SeqCst) {
-				true => Ok(()),
-				false => std::future::pending().await,
-			}
+			std::future::poll_fn(|_| {
+				if self.stopped.load(Ordering::SeqCst) {
+					return Poll::Ready(Err(FakeError));
+				}
+				match self.finished.load(Ordering::SeqCst) && !self.never_ack.load(Ordering::SeqCst) {
+					true => Poll::Ready(Ok(())),
+					false => Poll::Pending,
+				}
+			})
+			.await
 		}
 	}
 
@@ -931,7 +926,7 @@ mod tests {
 		let mut send = SendStream::new(fake.clone());
 		let mut cx = cx();
 
-		// A pre-terminal closed watch stays pending without consuming the stream.
+		// An idle closed watch stays pending until the peer cancels.
 		assert!(send.poll_closed(&mut cx).is_pending());
 
 		// The write proceeds; the old ownership-transfer watch deadlocked here.
@@ -939,6 +934,16 @@ mod tests {
 		assert_eq!(fake.writes.lock().unwrap().as_slice(), b"hello");
 	}
 
+	#[test]
+	fn send_closed_observes_stop_while_idle() {
+		let fake = FakeSend::default();
+		let mut send = SendStream::new(fake.clone());
+		let mut cx = cx();
+		assert!(send.poll_closed(&mut cx).is_pending());
+		fake.stopped.store(true, Ordering::SeqCst);
+		assert!(matches!(send.poll_closed(&mut cx), Poll::Ready(Err(_))));
+	}
+
 	// After finish() the real closed() watch runs and resolves.
 	#[test]
 	fn send_closed_resolves_after_finish() {