diff --git a/packages/svelte/src/internal/client/reactivity/batch.js b/packages/svelte/src/internal/client/reactivity/batch.js
index 792f84ad88..281e6fa0ff 100644
--- a/packages/svelte/src/internal/client/reactivity/batch.js
+++ b/packages/svelte/src/internal/client/reactivity/batch.js
@@ -285,6 +285,13 @@ export class Batch {
 	 */
 	#scheduled = [];
 
+	/**
+	 * Effects scheduled outside of a flush, and the status they are to have. Their status is only
+	 * applied when this batch is processed, so that other batches don't run them before that
+	 * @type {Map<Effect, number>}
+	 */
+	#unprocessed = new Map();
+
 	/**
 	 * Deferred reactions and their status.
 	 *
@@ -402,13 +409,11 @@ export class Batch {
 			this.#skipped_branches.delete(effect);
 
 			for (var e of tracked.d) {
-				set_signal_status(e, DIRTY);
-				this.schedule(e);
+				this.schedule(e, DIRTY);
 			}
 
 			for (e of tracked.m) {
-				set_signal_status(e, MAYBE_DIRTY);
-				this.schedule(e);
+				this.schedule(e, MAYBE_DIRTY);
 			}
 		}
 		this.unskipped_branches.add(effect);
@@ -441,8 +446,7 @@ export class Batch {
 			// TODO maybe we find a way to instead find the (pending) async effect and set the resulting value on this batch
 			(effect.deps !== null || (effect.f & (REACTION_RAN | ASYNC)) !== REACTION_RAN)
 		) {
-			set_signal_status(effect, DIRTY);
-			this.schedule(effect);
+			this.schedule(effect, DIRTY);
 
 			// The effect's state will reflect this batch, so other forks that ran it need to re-run it.
 			// Not so for async effects, as their result is stored in the fork and remains valid.
@@ -524,12 +528,17 @@ export class Batch {
 		for (const [reaction, status] of this.#dirty_reactions) {
 			if ((reaction.f & DERIVED) !== 0) {
 				set_signal_status(reaction, status);
-			} else if (status === DIRTY || (reaction.f & DIRTY) === 0) {
-				set_signal_status(reaction, status);
-				this.schedule(/** @type {Effect} */ (reaction));
+			} else {
+				this.schedule(/** @type {Effect} */ (reaction), status);
 			}
 		}
 
+		for (const [effect, status] of this.#unprocessed) {
+			this.#add(effect, status);
+		}
+
+		this.#unprocessed.clear();
+
 		this.apply();
 
 		/** @type {Effect[]} */
@@ -575,7 +584,8 @@ export class Batch {
 		if (updates.length > 0) {
 			var batch = Batch.ensure();
 			for (const e of updates) {
-				batch.schedule(e);
+				// these were marked during traversal already (see `mark_reactions`), unless they ran since
+				if ((e.f & CLEAN) === 0) batch.schedule(e, e.f & (DIRTY | MAYBE_DIRTY));
 			}
 		}
 
@@ -633,6 +643,7 @@ export class Batch {
 		// Edge case: During traversal new branches might create effects that run immediately and set state,
 		// causing an effect to be scheduled again. We need to traverse the current batch
 		// once more in that case - most of the time this will just clean up dirty branches.
+		// TODO I think we can delete this now since we re-iterate above
 		if (this.#scheduled.length > 0) {
 			if (next_batch !== null) {
 				for (const e of this.#scheduled) {
@@ -767,8 +778,7 @@ export class Batch {
 					if (this.#dirty_reactions.get(effect) === MAYBE_DIRTY) {
 						this.#dirty_reactions.delete(effect);
 					}
-					set_signal_status(effect, status);
-					this.schedule(effect);
+					this.schedule(effect, status);
 					marked = true;
 				}
 			}
@@ -1239,12 +1249,39 @@ export class Batch {
 	}
 
 	/**
-	 *
+	 * Schedule `effect` to run in this batch. Outside of a flush (and of effects running
+	 * synchronously, e.g. during mount), the status is kept to this batch until it is processed, so that other
+	 * batches don't run the effect (and mark it clean) before that, and so that other batches scheduling it don't
+	 * skip it as already dirty. During a flush it is applied right away, so that the ongoing flush can run the effect
 	 * @param {Effect} effect
+	 * @param {number} status `DIRTY` or `MAYBE_DIRTY`
 	 */
-	schedule(effect) {
+	schedule(effect, status) {
 		last_scheduled_effect = effect;
 
+		if (
+			is_processing ||
+			active_reaction !== null ||
+			// fast path to avoid map lookups when there's only one batch (no danger of other batches stealing this batch's effects)
+			(!this.is_fork && !(/** @type {Batch} */ (first_batch).next))
+		) {
+			this.#add(effect, status);
+		} else if (status === DIRTY || this.#unprocessed.get(effect) !== DIRTY) {
+			this.#unprocessed.set(effect, status);
+		}
+	}
+
+	/**
+	 * Apply the status to `effect` and add it to the effects that this batch is going to process
+	 * @param {Effect} effect
+	 * @param {number} status
+	 */
+	#add(effect, status) {
+		// don't set a DIRTY effect to MAYBE_DIRTY
+		if ((effect.f & DIRTY) === 0) {
+			set_signal_status(effect, status);
+		}
+
 		// defer render effects inside a pending boundary
 		// TODO the `REACTION_RAN` check is only necessary because of legacy `$:` effects AFAICT — we can remove later
 		if (
@@ -1253,10 +1290,9 @@ export class Batch {
 			(effect.f & REACTION_RAN) === 0
 		) {
 			effect.b.defer_effect(effect);
-			return;
+		} else {
+			this.#scheduled.push(effect);
 		}
-
-		this.#scheduled.push(effect);
 	}
 
 	#unlink() {
@@ -1436,10 +1472,11 @@ function flush_queued_effects(effects) {
 
 /**
  * @param {Effect} effect
+ * @param {number} status `DIRTY` or `MAYBE_DIRTY`
  * @returns {void}
  */
-export function schedule_effect(effect) {
-	/** @type {Batch} */ (current_batch).schedule(effect);
+export function schedule_effect(effect, status) {
+	/** @type {Batch} */ (current_batch).schedule(effect, status);
 }
 
 /** @type {Source<number>[]} */
diff --git a/packages/svelte/src/internal/client/reactivity/effects.js b/packages/svelte/src/internal/client/reactivity/effects.js
index 10cebec1fe..abe38fafb0 100644
--- a/packages/svelte/src/internal/client/reactivity/effects.js
+++ b/packages/svelte/src/internal/client/reactivity/effects.js
@@ -130,7 +130,7 @@ function create_effect(type, fn) {
 			collected_effects.push(effect);
 		} else {
 			// schedule for later
-			Batch.ensure().schedule(effect);
+			Batch.ensure().schedule(effect, DIRTY);
 		}
 	} else if (fn !== null) {
 		try {
@@ -707,8 +707,7 @@ function resume_children(effect, local) {
 	// here because we don't want to eagerly recompute a derived like
 	// `{#if foo}{foo.bar()}{/if}` if `foo` is now `undefined
 	if ((effect.f & CLEAN) === 0) {
-		set_signal_status(effect, DIRTY);
-		Batch.ensure().schedule(effect); // Assumption: This happens during the commit phase of the batch, causing another flush, but it's safe
+		Batch.ensure().schedule(effect, DIRTY); // Assumption: This happens during the commit phase of the batch, causing another flush, but it's safe
 	}
 
 	var child = effect.first;
diff --git a/packages/svelte/src/internal/client/reactivity/sources.js b/packages/svelte/src/internal/client/reactivity/sources.js
index b001702093..5fb761eeec 100644
--- a/packages/svelte/src/internal/client/reactivity/sources.js
+++ b/packages/svelte/src/internal/client/reactivity/sources.js
@@ -357,15 +357,14 @@ export function increment(source) {
  * @param {Reaction} reaction
  */
 export function invalidate(reaction) {
-	set_signal_status(reaction, DIRTY);
-
 	if ((reaction.f & DERIVED) !== 0) {
+		set_signal_status(reaction, DIRTY);
 		seen = null;
 		count_deps = 0;
 		mark_reactions(/** @type {Derived} */ (reaction), DIRTY, null);
 		seen = null;
 	} else {
-		schedule_effect(/** @type {Effect} */ (reaction));
+		schedule_effect(/** @type {Effect} */ (reaction), DIRTY);
 	}
 }
 
@@ -403,8 +402,12 @@ function mark_reactions(signal, status, updated_during_traversal) {
 
 		var not_dirty = (flags & DIRTY) === 0;
 
-		// don't set a DIRTY reaction to MAYBE_DIRTY
-		if (not_dirty) {
+		// don't set a DIRTY reaction to MAYBE_DIRTY. Scheduled effects get
+		// their status from the batch they're scheduled in (see `Batch#schedule`)
+		if (
+			not_dirty &&
+			((flags & (EAGER_EFFECT | DERIVED)) !== 0 || updated_during_traversal !== null)
+		) {
 			set_signal_status(reaction, status);
 		}
 
@@ -427,7 +430,7 @@ function mark_reactions(signal, status, updated_during_traversal) {
 			if (updated_during_traversal !== null) {
 				updated_during_traversal.push(effect);
 			} else {
-				schedule_effect(effect);
+				schedule_effect(effect, status);
 			}
 		}
 	}
diff --git a/packages/svelte/src/internal/client/runtime.js b/packages/svelte/src/internal/client/runtime.js
index 85883e0076..7518b11742 100644
--- a/packages/svelte/src/internal/client/runtime.js
+++ b/packages/svelte/src/internal/client/runtime.js
@@ -222,12 +222,7 @@ function schedule_possible_effect_self_invalidation(signal, effect, root = true)
 		if ((reaction.f & DERIVED) !== 0) {
 			schedule_possible_effect_self_invalidation(/** @type {Derived} */ (reaction), effect, false);
 		} else if (effect === reaction) {
-			if (root) {
-				set_signal_status(reaction, DIRTY);
-			} else if ((reaction.f & CLEAN) !== 0) {
-				set_signal_status(reaction, MAYBE_DIRTY);
-			}
-			schedule_effect(/** @type {Effect} */ (reaction));
+			schedule_effect(/** @type {Effect} */ (reaction), root ? DIRTY : MAYBE_DIRTY);
 		}
 	}
 }
@@ -512,7 +507,7 @@ export function update_effect(effect) {
 			// dirty, since their results are kept per batch, and they run during traversal: marking them dirty
 			// would re-run them (and restart async work) on every process of a pending batch.
 			var own = /** @type {Batch} */ (own_batch);
-			var status = (flags & (EFFECT | RENDER_EFFECT | MANAGED_EFFECT)) !== 0 ? DIRTY : MAYBE_DIRTY;
+			var status = (flags & (BLOCK_EFFECT | ASYNC)) !== 0 ? MAYBE_DIRTY : DIRTY;
 			own.stale_effects.set(effect, write_version);
 			for (var batch = own.next; batch !== null && effect.deps !== null; batch = batch.next) {
 				batch.add_dirty_reaction(effect, status);
diff --git a/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/Sync.svelte b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/Sync.svelte
new file mode 100644
index 0000000000..eee3e94e31
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/Sync.svelte
@@ -0,0 +1,5 @@
+<script>
+	let { c } = $props();
+</script>
+
+<span>{c}</span>
diff --git a/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/_config.js b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/_config.js
new file mode 100644
index 0000000000..6c6f8100e5
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/_config.js
@@ -0,0 +1,28 @@
+import { tick } from 'svelte';
+import { test } from '../../test';
+
+// An effect that several batches scheduled must run in each of them, even if another batch is flushed first
+export default test({
+	mode: ['client'],
+	async test({ assert, target }) {
+		await tick();
+		const [a, b, resolve_and_write, resolve] = target.querySelectorAll('button');
+		const buttons =
+			'<button>a</button><button>b</button><button>resolve and write</button><button>resolve</button>';
+		assert.htmlEqual(target.innerHTML, `${buttons}<p>0 0</p><i>0</i><span>0</span>`);
+
+		a.click();
+		await tick();
+		b.click();
+		await tick();
+		assert.htmlEqual(target.innerHTML, `${buttons}<p>0 0</p><i>0</i><span>0</span>`);
+
+		resolve_and_write.click();
+		await tick();
+		assert.htmlEqual(target.innerHTML, `${buttons}<p>0 1</p><i>0</i><span>1</span>`);
+
+		resolve.click();
+		await tick();
+		assert.htmlEqual(target.innerHTML, `${buttons}<p>1 1</p><i>1</i><span>1</span>`);
+	}
+});
diff --git a/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/main.svelte b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/main.svelte
new file mode 100644
index 0000000000..18ff344f79
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects-multiple/main.svelte
@@ -0,0 +1,33 @@
+<script>
+	import Sync from './Sync.svelte';
+
+	let a = $state(0);
+	let b = $state(0);
+	let c = $state(0);
+
+	const registry = [];
+
+	function load(value) {
+		if (!value) return value;
+		return new Promise((resolve) => registry.push(() => resolve(value)));
+	}
+</script>
+
+<button onclick={() => a++}>a</button>
+<button onclick={() => b++}>b</button>
+<button
+	onclick={() => {
+		// the `<p>` values of both batches resolve into their batches (scheduling the same effect in each
+		// of them) after this third batch was created, but before it is flushed. The first batch stays pending
+		const [p_a, i_a, p_b] = registry.splice(0);
+		registry.push(i_a);
+		p_b();
+		p_a();
+		queueMicrotask(() => c++);
+	}}>resolve and write</button
+>
+<button onclick={() => registry.shift()?.()}>resolve</button>
+
+<p>{await load(a)} {await load(b)}</p>
+<i>{await load(a)}</i>
+<Sync {c} />
diff --git a/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/Async.svelte b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/Async.svelte
new file mode 100644
index 0000000000..f24b8d26a3
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/Async.svelte
@@ -0,0 +1,6 @@
+<script>
+	let { x } = $props();
+	const result = $derived(await x);
+</script>
+
+<p>block: {result}</p>
diff --git a/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/Sync.svelte b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/Sync.svelte
new file mode 100644
index 0000000000..057e1db394
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/Sync.svelte
@@ -0,0 +1,5 @@
+<script>
+	let { y } = $props();
+</script>
+
+<span>{y}</span>
diff --git a/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/_config.js b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/_config.js
new file mode 100644
index 0000000000..5f860cf2e6
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/_config.js
@@ -0,0 +1,16 @@
+import { tick } from 'svelte';
+import { test } from '../../test';
+
+// An effect must run in the batch that scheduled it, not in another batch that happens to be flushed first
+export default test({
+	mode: ['client'],
+	async test({ assert, target }) {
+		await tick();
+		const [button] = target.querySelectorAll('button');
+		assert.htmlEqual(target.innerHTML, '<button>increment</button><p>block: 0</p><span>0</span>');
+
+		button.click();
+		await tick();
+		assert.htmlEqual(target.innerHTML, '<button>increment</button><p>block: 1</p><span>1</span>');
+	}
+});
diff --git a/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/main.svelte b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/main.svelte
new file mode 100644
index 0000000000..b40ff30317
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-batch-flush-stolen-effects/main.svelte
@@ -0,0 +1,18 @@
+<script>
+	import Async from './Async.svelte';
+	import Sync from './Sync.svelte';
+
+	let x = $state(0);
+	let y = $state(0);
+</script>
+
+<button
+	onclick={() => {
+		x++;
+		// the second batch is created while the first one is pending, after the first one
+		// scheduled its effects but before it is flushed again
+		queueMicrotask(() => queueMicrotask(() => y++));
+	}}>increment</button
+>
+<Async {x} />
+<Sync {y} />
diff --git a/packages/svelte/tests/runtime-runes/samples/async-fork-flush-stolen-effects-render/_config.js b/packages/svelte/tests/runtime-runes/samples/async-fork-flush-stolen-effects-render/_config.js
new file mode 100644
index 0000000000..4a3c821f47
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-fork-flush-stolen-effects-render/_config.js
@@ -0,0 +1,29 @@
+import { tick } from 'svelte';
+import { test } from '../../test';
+
+// A fork flush defers the render effects it comes across (marking them clean), and resets the ones
+// inside branches it is going to remove. Neither may hide them from the real batch that scheduled them
+export default test({
+	mode: ['client'],
+	async test({ assert, target }) {
+		const [fork, resolve, resolve_and_write, discard] = target.querySelectorAll('button');
+		const buttons =
+			'<button>fork</button><button>resolve</button><button>resolve and write</button><button>discard</button>';
+
+		resolve.click();
+		await tick();
+		assert.htmlEqual(target.innerHTML, `${buttons}<span>0</span><em>0</em><q>0</q>`);
+
+		fork.click();
+		await tick();
+		assert.htmlEqual(target.innerHTML, `${buttons}<span>0</span><em>0</em><q>0</q>`);
+
+		resolve_and_write.click();
+		await tick();
+		assert.htmlEqual(target.innerHTML, `${buttons}<span>1</span><em>1</em><q>0</q>`);
+
+		discard.click();
+		await tick();
+		assert.htmlEqual(target.innerHTML, `${buttons}<span>1</span><em>1</em><q>0</q>`);
+	}
+});
diff --git a/packages/svelte/tests/runtime-runes/samples/async-fork-flush-stolen-effects-render/main.svelte b/packages/svelte/tests/runtime-runes/samples/async-fork-flush-stolen-effects-render/main.svelte
new file mode 100644
index 0000000000..ea61e3c4f6
--- /dev/null
+++ b/packages/svelte/tests/runtime-runes/samples/async-fork-flush-stolen-effects-render/main.svelte
@@ -0,0 +1,35 @@
+<script>
+	import { fork } from 'svelte';
+
+	let a = $state(0);
+	let b = $state(0);
+	let show = $state(true);
+
+	const registry = [];
+	let f;
+
+	function load(value) {
+		return new Promise((resolve) => registry.push(() => resolve(value)));
+	}
+</script>
+
+<button
+	onclick={() =>
+		(f = fork(() => {
+			b = 1;
+			show = false;
+		}))}>fork</button
+>
+<button onclick={() => registry.shift()?.()}>resolve</button>
+<button
+	onclick={() => {
+		// the fork's promise resolves (and the fork is flushed) before the real batch is
+		registry.shift()?.();
+		a += 1;
+	}}>resolve and write</button
+>
+<button onclick={() => f.discard()}>discard</button>
+
+<span>{a}</span>
+{#if show}<em>{a}</em>{/if}
+{#await load(b) then v}<q>{v}</q>{/await}