diff --git a/.changeset/pre/live-query-stream-teardown.md b/.changeset/pre/live-query-stream-teardown.md
new file mode 100644
index 000000000000..7e6667abb7a6
--- /dev/null
+++ b/.changeset/pre/live-query-stream-teardown.md
@@ -0,0 +1,5 @@
+---
+'@sveltejs/kit': patch
+---
+
+fix: don't touch the `query.live` stream controller after teardown, and make response cancellation observable via the generator's `request.signal`
diff --git a/packages/kit/src/runtime/server/remote-functions.js b/packages/kit/src/runtime/server/remote-functions.js
index 43674bc0abf9..0b76701e2394 100644
--- a/packages/kit/src/runtime/server/remote-functions.js
+++ b/packages/kit/src/runtime/server/remote-functions.js
@@ -1,7 +1,7 @@
 /** @import { RequestEvent, SSRManifest } from '@sveltejs/kit' */
 /** @import { RemoteForm } from '$app/server' */
 /** @import { ActionResult } from '$app/forms' */
-/** @import { RemoteFormInternals, RemoteFunctionData, RemoteFunctionResponse, RemoteInternals, RequestState, SSROptions } from 'types' */
+/** @import { RemoteFormInternals, RemoteFunctionData, RemoteFunctionResponse, RemoteInternals, RemoteQueryLiveInternals, RequestState, SSROptions } from 'types' */
 
 import { json, error } from '@sveltejs/kit';
 import { Redirect, SvelteKitError } from '@sveltejs/kit/internal';
@@ -15,7 +15,7 @@ import { normalize_error } from '../../utils/error.js';
 import { check_incorrect_fail_use, get_action_location } from './page/actions.js';
 import { DEV } from 'esm-env';
 import { deserialize_binary_form } from '../form-utils.js';
-import { with_version_header } from './utils.js';
+import { stream_from_iterator, with_version_header } from './utils.js';
 
 /**
  * How long (in milliseconds) to wait after the last message was sent before
@@ -24,6 +24,8 @@ import { with_version_header } from './utils.js';
  */
 const KEEP_ALIVE_INTERVAL = 30_000;
 
+const KEEP_ALIVE = Symbol('keep-alive');
+
 /** @type {typeof handle_remote_call_internal} */
 export async function handle_remote_call(event, state, options, manifest, id) {
 	return record_span({
@@ -42,13 +44,11 @@ export async function handle_remote_call(event, state, options, manifest, id) {
 }
 
 /**
- * @param {RequestEvent} event
- * @param {RequestState} state
- * @param {SSROptions} options
+ * Looks a remote function up in the manifest by its request id.
  * @param {SSRManifest} manifest
  * @param {string} id
  */
-async function handle_remote_call_internal(event, state, options, manifest, id) {
+async function resolve_remote_function(manifest, id) {
 	const [hash, name, additional_args] = id.split('/');
 	const remotes = manifest._.remotes;
 
@@ -59,149 +59,168 @@ async function handle_remote_call_internal(event, state, options, manifest, id)
 
 	if (!fn) error(404);
 
-	/** @type {RemoteInternals} */
-	const internals = fn.__;
+	return { fn, internals: /** @type {RemoteInternals} */ (fn.__), additional_args };
+}
 
-	event.tracing.current.setAttributes({
-		'sveltekit.remote.call.type': internals.type,
-		'sveltekit.remote.call.name': internals.name
-	});
+/**
+ * @param {RemoteFunctionData} data
+ * @param {HeadersInit | undefined} headers
+ */
+function result_response(data, headers) {
+	return json(
+		/** @type {RemoteFunctionResponse} */ ({
+			type: 'result',
+			data: stringify(data)
+		}),
+		{ headers }
+	);
+}
 
-	/** @type {HeadersInit | undefined} */
-	const headers = state.prerendering ? undefined : { 'cache-control': 'private, no-store' };
+/**
+ * Handles a `query.live` call: runs the generator and streams its values as
+ * server-sent events.
+ * @param {RequestEvent} event
+ * @param {RequestState} state
+ * @param {SSROptions} options
+ * @param {RemoteQueryLiveInternals} internals
+ */
+function handle_live_query(event, state, options, internals) {
+	if (event.request.method !== 'GET') {
+		throw new SvelteKitError(
+			405,
+			'Method Not Allowed',
+			`\`query.live\` functions must be invoked via GET request, not ${event.request.method}`
+		);
+	}
 
-	try {
-		/** @type {RemoteFunctionData} */
-		const data = {};
+	const payload = /** @type {string} */ (new URL(event.request.url).searchParams.get('payload'));
 
-		switch (internals.type) {
-			case 'query_live': {
-				if (event.request.method !== 'GET') {
-					throw new SvelteKitError(
-						405,
-						'Method Not Allowed',
-						`\`query.live\` functions must be invoked via GET request, not ${event.request.method}`
-					);
-				}
+	// aborted whenever the stream is torn down, so unlike the request signal it
+	// also fires on response teardown, which the generator could otherwise
+	// never observe
+	const cancellation = new AbortController();
 
-				const payload = /** @type {string} */ (
-					new URL(event.request.url).searchParams.get('payload')
-				);
+	if (event.request.signal.aborted) {
+		cancellation.abort();
+	} else {
+		event.request.signal.addEventListener('abort', () => cancellation.abort(), {
+			once: true
+		});
+	}
 
-				const generator = internals.run(event, state, parse_remote_arg(payload));
-
-				const encoder = new TextEncoder();
-
-				let closed = false;
-
-				/** @type {ReturnType<typeof setTimeout> | undefined} */
-				let keep_alive;
-
-				/**
-				 * (Re)schedule the keep-alive comment. Called whenever a message is sent, so
-				 * that a keep-alive is only emitted once `KEEP_ALIVE_INTERVAL` has elapsed
-				 * without any other activity.
-				 * @param {ReadableStreamDefaultController} controller
-				 */
-				function schedule_keep_alive(controller) {
-					clearTimeout(keep_alive);
-					keep_alive = setTimeout(() => {
-						if (closed || event.request.signal.aborted) return;
-						// SSE comments (lines starting with `:`) are ignored by the client
-						controller.enqueue(encoder.encode(': keep-alive\n\n'));
-						schedule_keep_alive(controller);
-					}, KEEP_ALIVE_INTERVAL);
-				}
+	const live_event = {
+		...event,
+		request: new Request(event.request, { signal: cancellation.signal })
+	};
 
-				/**
-				 * @param {ReadableStreamDefaultController} controller
-				 * @param {any} payload
-				 */
-				function send(controller, payload) {
-					controller.enqueue(encoder.encode('data: ' + JSON.stringify(payload) + '\n\n'));
-					schedule_keep_alive(controller);
+	const generator = internals.run(live_event, state, parse_remote_arg(payload));
+
+	/** @param {any} payload */
+	const frame = (payload) => 'data: ' + JSON.stringify(payload) + '\n\n';
+
+	/**
+	 * Resolves with the next iterator result, or with `KEEP_ALIVE` once
+	 * `KEEP_ALIVE_INTERVAL` has elapsed without one.
+	 * @param {Promise<IteratorResult<any>>} pending
+	 * @returns {Promise<IteratorResult<any> | typeof KEEP_ALIVE>}
+	 */
+	function next_or_keep_alive(pending) {
+		return new Promise((resolve, reject) => {
+			const timer = setTimeout(() => resolve(KEEP_ALIVE), KEEP_ALIVE_INTERVAL);
+			pending.then(
+				(result) => {
+					clearTimeout(timer);
+					resolve(result);
+				},
+				(error) => {
+					clearTimeout(timer);
+					reject(error);
 				}
+			);
+		});
+	}
+
+	/** @type {string | undefined} */
+	let result = undefined;
 
-				/** @type {string | undefined} */
-				let result = undefined;
+	// everything the stream sends, as a generator of SSE strings — it holds no
+	// reference to the stream controller, so it cannot touch a dead one
+	async function* frames() {
+		/** @type {Promise<IteratorResult<any>> | null} */
+		let pending = null;
 
-				async function cancel() {
-					if (closed) return;
-					closed = true;
-					clearTimeout(keep_alive);
-					await generator.return(undefined);
+		try {
+			while (true) {
+				pending ??= generator.next();
+				const winner = await next_or_keep_alive(pending);
+
+				if (winner === KEEP_ALIVE) {
+					// SSE comments (lines starting with `:`) are ignored by the client
+					yield ': keep-alive\n\n';
+					continue;
 				}
 
-				event.request.signal.addEventListener('abort', cancel, { once: true });
+				pending = null;
 
-				return new Response(
-					new ReadableStream({
-						start(controller) {
-							schedule_keep_alive(controller);
-						},
-						async pull(controller) {
-							if (event.request.signal.aborted) {
-								await cancel();
-								controller.close();
-								return;
-							}
+				if (winner.done) return;
 
-							try {
-								while (true) {
-									const { value, done } = await generator.next();
-
-									if (done) {
-										await cancel();
-										controller.close();
-										return;
-									}
-
-									// only send changed data
-									if (result !== (result = stringify(value))) {
-										send(controller, {
-											type: 'result',
-											result
-										});
-
-										return;
-									}
-								}
-							} catch (error) {
-								if (!event.request.signal.aborted) {
-									if (error instanceof Redirect) {
-										send(controller, {
-											type: 'redirect',
-											location: error.location
-										});
-									} else {
-										const transformed = await handle_error_and_jsonify(
-											event,
-											state,
-											options,
-											error
-										);
-
-										send(controller, {
-											type: 'error',
-											error: transformed
-										});
-									}
-								}
-
-								await cancel();
-								controller.close();
-							}
-						},
-						cancel
-					}),
-					{
-						headers: {
-							'cache-control': 'private, no-store',
-							'content-type': 'text/event-stream'
-						}
-					}
-				);
+				// only send changed data
+				if (result !== (result = stringify(winner.value))) {
+					yield frame({ type: 'result', result });
+				}
 			}
+		} catch (error) {
+			if (!cancellation.signal.aborted) {
+				if (error instanceof Redirect) {
+					yield frame({ type: 'redirect', location: error.location });
+				} else {
+					yield frame({
+						type: 'error',
+						error: await handle_error_and_jsonify(event, state, options, error)
+					});
+				}
+			}
+		} finally {
+			cancellation.abort();
+			await generator.return(undefined);
+		}
+	}
+
+	return new Response(
+		stream_from_iterator(frames(), () => cancellation.abort()),
+		{
+			headers: {
+				'cache-control': 'private, no-store',
+				'content-type': 'text/event-stream'
+			}
+		}
+	);
+}
+/**
+ * @param {RequestEvent} event
+ * @param {RequestState} state
+ * @param {SSROptions} options
+ * @param {SSRManifest} manifest
+ * @param {string} id
+ */
+async function handle_remote_call_internal(event, state, options, manifest, id) {
+	const { fn, internals, additional_args } = await resolve_remote_function(manifest, id);
+
+	event.tracing.current.setAttributes({
+		'sveltekit.remote.call.type': internals.type,
+		'sveltekit.remote.call.name': internals.name
+	});
+
+	/** @type {HeadersInit | undefined} */
+	const headers = state.prerendering ? undefined : { 'cache-control': 'private, no-store' };
+
+	try {
+		/** @type {RemoteFunctionData} */
+		const data = {};
+
+		switch (internals.type) {
+			case 'query_live':
+				return handle_live_query(event, state, options, internals);
 
 			case 'query_batch': {
 				if (event.request.method !== 'POST') {
@@ -262,13 +281,7 @@ async function handle_remote_call_internal(event, state, options, manifest, id)
 
 				if (data._.issues) {
 					// special case — don't serialize refreshes/reconnects
-					return json(
-						/** @type {RemoteFunctionResponse} */ ({
-							type: 'result',
-							data: stringify(data)
-						}),
-						{ headers }
-					);
+					return result_response(data, headers);
 				}
 
 				break;
@@ -310,24 +323,12 @@ async function handle_remote_call_internal(event, state, options, manifest, id)
 
 		await collect_remote_data(data, event, state, options);
 
-		return json(
-			/** @type {RemoteFunctionResponse} */ ({
-				type: 'result',
-				data: stringify(data)
-			}),
-			{ headers }
-		);
+		return result_response(data, headers);
 	} catch (error) {
 		if (error instanceof Redirect) {
 			const data = await collect_remote_data({ redirect: error.location }, event, state, options);
 
-			return json(
-				/** @type {RemoteFunctionResponse} */ ({
-					type: 'result',
-					data: stringify(data)
-				}),
-				{ headers }
-			);
+			return result_response(data, headers);
 		}
 
 		const transformed = await handle_error_and_jsonify(event, state, options, error);
diff --git a/packages/kit/src/runtime/server/utils.js b/packages/kit/src/runtime/server/utils.js
index 990decb06821..67105a351655 100644
--- a/packages/kit/src/runtime/server/utils.js
+++ b/packages/kit/src/runtime/server/utils.js
@@ -1,5 +1,39 @@
 import { text } from '@sveltejs/kit';
 import { ENDPOINT_METHODS } from '../../constants.js';
+import { text_encoder } from '../utils.js';
+
+/**
+ * Builds a text stream from an iterator of string chunks. The controller is
+ * confined here so that a chunk arriving after the stream was torn down is
+ * dropped instead of hitting a closed controller — either side can tear the
+ * stream down while `pull` is suspended on `iterator.next()`.
+ * @param {AsyncIterator<string>} iterator
+ * @param {() => void} [oncancel] called when the consumer cancels the stream
+ * @returns {ReadableStream<Uint8Array>}
+ */
+export function stream_from_iterator(iterator, oncancel) {
+	let open = true;
+
+	return new ReadableStream({
+		async pull(controller) {
+			const { value, done } = await iterator.next();
+
+			if (!open) return;
+
+			if (done) {
+				open = false;
+				controller.close();
+			} else {
+				controller.enqueue(text_encoder.encode(value));
+			}
+		},
+		async cancel() {
+			open = false;
+			oncancel?.();
+			await iterator.return?.(undefined);
+		}
+	});
+}
 
 /**
  * @param {Partial<Record<import('types').HttpMethod, any>>} mod
diff --git a/packages/kit/src/runtime/server/utils.spec.js b/packages/kit/src/runtime/server/utils.spec.js
new file mode 100644
index 000000000000..7f12a67fa81a
--- /dev/null
+++ b/packages/kit/src/runtime/server/utils.spec.js
@@ -0,0 +1,76 @@
+import { expect, test, vi } from 'vitest';
+import { stream_from_iterator } from './utils.js';
+
+const decoder = new TextDecoder();
+
+/** @param {string[]} chunks */
+async function* from(chunks) {
+	for (const chunk of chunks) {
+		await Promise.resolve();
+		yield chunk;
+	}
+}
+
+test('streams encoded chunks and closes when the iterator is done', async () => {
+	const reader = stream_from_iterator(from(['one', 'two'])).getReader();
+
+	const first = await reader.read();
+	expect(decoder.decode(first.value)).toBe('one');
+
+	const second = await reader.read();
+	expect(decoder.decode(second.value)).toBe('two');
+
+	await expect(reader.read()).resolves.toEqual({ value: undefined, done: true });
+});
+
+test('cancellation notifies the producer and returns the iterator', async () => {
+	let finished = false;
+	const oncancel = vi.fn();
+
+	async function* source() {
+		try {
+			await Promise.resolve();
+			yield 'one';
+			yield 'two';
+		} finally {
+			finished = true;
+		}
+	}
+
+	const reader = stream_from_iterator(source(), oncancel).getReader();
+	await reader.read();
+
+	await reader.cancel();
+
+	expect(oncancel).toHaveBeenCalledOnce();
+	expect(finished).toBe(true);
+});
+
+// https://github.com/sveltejs/kit/issues/16778
+test('cancellation settles while the producer is parked', async () => {
+	/** @type {(result: IteratorResult<string>) => void} */
+	let resolve_next = () => {};
+	let returned = false;
+
+	/** @type {AsyncIterator<string>} */
+	const iterator = {
+		next: () => new Promise((resolve) => (resolve_next = resolve)),
+		return: () => {
+			returned = true;
+			return Promise.resolve({ value: undefined, done: true });
+		}
+	};
+
+	const reader = stream_from_iterator(iterator).getReader();
+	const read = reader.read(); // pull is now suspended on `iterator.next()`
+	await Promise.resolve();
+
+	// neither cancel() nor the pending read may wait for the parked next()
+	await reader.cancel();
+	expect(returned).toBe(true);
+	await expect(read).resolves.toEqual({ value: undefined, done: true });
+
+	// the parked value landing afterwards is a no-op
+	resolve_next({ value: 'late', done: false });
+	await new Promise((resolve) => setTimeout(resolve, 0));
+});