diff --git a/param/reactive.py b/param/reactive.py
index 05229f35..bf7eb74f 100644
--- a/param/reactive.py
+++ b/param/reactive.py
@@ -1411,6 +1411,27 @@ def _remove_watcher(
         pass
 
 
+async def _close_stale(obj):
+    """
+    Discard an awaitable or async generator whose result is no longer needed.
+
+    Closing a coroutine that was never awaited also suppresses the warning
+    Python emits when it is garbage collected.
+    """
+    try:
+        if inspect.isasyncgen(obj):
+            await obj.aclose()
+        elif inspect.iscoroutine(obj):
+            obj.close()
+    except (StopAsyncIteration, GeneratorExit):
+        pass
+    except Exception:
+        logger.debug(
+            "Ignoring close error for stale reactive task.",
+            exc_info=True,
+        )
+
+
 # When we only support python >= 3.11 we should exchange 'rx' with Self type annotation below.
 # See https://peps.python.org/pep-0673/
 
@@ -1576,6 +1597,8 @@ def __init__(
         self._dirty_obj = False
         self._current_task = None
         self._resolve_generation = 0
+        self._finished_generation = 0
+        self._skipped = False
         self._error_state = None
         self._current_ = _current
         # _shared is used for branching rx pipelines where we clone the input.
@@ -1679,6 +1702,14 @@ def _is_async(self) -> bool:
             inspect.isgeneratorfunction(fn)
         )
 
+    @property
+    def _awaiting(self) -> bool:
+        """
+        Whether an asynchronous resolution is in flight that has not yet
+        produced a value for the current generation.
+        """
+        return self._resolve_generation != self._finished_generation
+
     @property
     def _current(self):
         if self._error_state:
@@ -1794,14 +1825,23 @@ def _invalidate_obj(self, *events):
         t.cast('t.Any', self._root)._dirty_obj = True
         self._error_state = None
 
-    async def _resolve_async(self, obj=None, generation=None):
+    async def _resolve_async(self, obj=None, generation: int = 0):
         import asyncio
-        self._current_task = task = asyncio.current_task()
-        trigger = self._trigger
 
         def stale():
             return generation != self._resolve_generation
 
+        if stale():
+            # A newer resolution was requested before this task was scheduled,
+            # so nothing has awaited obj yet and the operation has not begun.
+            # Close it instead of computing a result that is already superseded.
+            # This must happen before _current_task is claimed below, otherwise
+            # the finally clause would clear the genuinely current task and hide
+            # it from the next _lazy_resolve.
+            await _close_stale(obj)
+            return
+        self._current_task = task = asyncio.current_task()
+        trigger = self._trigger
         try:
             if trigger is None:
                 return
@@ -1814,27 +1854,22 @@ def stale():
                 if stale():
                     return
                 self._current_ = shared.rx.value
+                self._finished_generation = generation
                 trigger.param.trigger('value')
             elif inspect.isasyncgen(obj):
                 async for val in obj:
                     if stale():
-                        try:
-                            await obj.aclose()
-                        except (StopAsyncIteration, GeneratorExit):
-                            pass
-                        except Exception:
-                            logger.debug(
-                                "Ignoring async generator close error for stale reactive task.",
-                                exc_info=True,
-                            )
+                        await _close_stale(obj)
                         break
                     self._current_ = val
+                    self._finished_generation = generation
                     trigger.param.trigger('value')
             else:
                 value = await obj
                 if stale():
                     return
                 self._current_ = value
+                self._finished_generation = generation
                 trigger.param.trigger('value')
         except asyncio.CancelledError:
             return
@@ -1862,6 +1897,8 @@ def _resolve(self):
                 if obj is Skip or obj is Undefined:
                     self._current_ = Undefined
                     raise Skip
+                elif self._prev is not None and self._prev._skipped:
+                    raise Skip
                 elif (
                     self._shared is not None and
                     self._method is None and
@@ -1873,9 +1910,14 @@ def _resolve(self):
                     if self._is_async:
                         self._shared.rx.value # trigger async resolve
                         self._lazy_resolve()
-                    else:
-                        self._current_ = self._shared.rx.value
-                    raise Skip
+                        raise Skip
+                    # Returns instead of raising Skip because this path does
+                    # resolve to a value, so it must mirror the shared node's
+                    # skip state rather than be marked skipped by the handler.
+                    self._current_ = self._shared.rx.value
+                    self._skipped = self._shared._skipped
+                    self._dirty = False
+                    return self._current_
                 operation = self._operation
                 if operation:
                     obj = self._eval_operation(obj, operation)
@@ -1886,13 +1928,19 @@ def _resolve(self):
                         raise Skip
             except Skip:
                 self._dirty = False
+                self._skipped = True
                 return self._current_
             except Exception as e:
                 self._error_state = e
                 raise e
             self._current_ = current = obj
+            self._skipped = False
         else:
             current = self._current_
+            # A node awaiting an asynchronous result still holds the value it
+            # computed from the previous inputs; report it as skipped so it is
+            # not propagated as if it were current.
+            self._skipped = self._awaiting
         self._dirty = False
         if self._method:
             # E.g. `pi = dfi.A` leads to `pi._method` equal to `'A'`.
@@ -2232,6 +2280,11 @@ def __setattr__(self, name, value):
 def _rx_transform(obj):
     if not isinstance(obj, rx):
         return obj
-    return bind(lambda *_: obj.rx.value, *obj._params)
+    def resolve(*_):
+        value = obj.rx.value
+        if obj._skipped or value is Skip or value is Undefined:
+            raise Skip
+        return value
+    return bind(resolve, *obj._params)
 
 register_reference_transform(_rx_transform)
diff --git a/tests/testreactive.py b/tests/testreactive.py
index 0050eeca..3da645e7 100644
--- a/tests/testreactive.py
+++ b/tests/testreactive.py
@@ -866,6 +866,209 @@ async def gen(value, i):
     await async_wait_until(lambda: rxgen.rx.value == 10, interval=10)
     await async_wait_until(lambda: rxgen.rx.value == 11)
 
+async def mul_slowly(value):
+    await asyncio.sleep(0.02)
+    return value*2
+
+async def test_reactive_async_watcher_not_notified_while_awaiting():
+    irx = rx(1)
+    async_rx = irx.rx.pipe(mul_slowly)
+    items = []
+    async_rx.rx.watch(items.append)
+    assert async_rx.rx.value is param.Undefined
+    await async_wait_until(lambda: items == [2])
+    irx.rx.value = 2
+
+    # The awaited value has not resolved yet, so the watcher must not be
+    # notified with the value computed from the previous input.
+    assert items == [2]
+
+    await async_wait_until(lambda: items == [2, 4])
+
+async def test_reactive_async_downstream_watcher_not_notified_while_awaiting():
+    irx = rx(1)
+    downstream = irx.rx.pipe(mul_slowly) + 10
+    items = []
+    downstream.rx.watch(items.append)
+    assert downstream.rx.value is param.Undefined
+    await async_wait_until(lambda: items == [12])
+    irx.rx.value = 2
+    assert items == [12]
+    await async_wait_until(lambda: items == [12, 14])
+
+async def test_reactive_async_downstream_not_computed_while_awaiting():
+    computed = []
+    def add(value):
+        computed.append(value)
+        return value+10
+
+    irx = rx(1)
+    downstream = irx.rx.pipe(mul_slowly).rx.pipe(add)
+    downstream.rx.watch()
+    downstream.rx.value
+    await async_wait_until(lambda: computed == [2])
+    irx.rx.value = 2
+
+    # The downstream operation must not be applied to the superseded value.
+    assert computed == [2]
+
+    await async_wait_until(lambda: computed == [2, 4])
+
+async def test_reactive_async_rapid_updates_notify_once():
+    irx = rx(1)
+    async_rx = irx.rx.pipe(mul_slowly)
+    items = []
+    async_rx.rx.watch(items.append)
+    async_rx.rx.value
+    await async_wait_until(lambda: items == [2])
+    for value in (2, 3, 4):
+        irx.rx.value = value
+    await async_wait_until(lambda: items == [2, 8])
+    assert items == [2, 8]
+
+def test_reactive_skip_value_does_not_notify_watcher():
+    P = Parameters(integer=1)
+
+    def skip_values(v):
+        if v > 2:
+            raise Skip
+        return v+1
+
+    i = rx(P.param.integer).rx.pipe(skip_values)
+    items = []
+    i.rx.watch(items.append)
+    P.integer = 2
+    assert items == [3]
+
+    # The operation skipped, so the watcher must not be notified with the
+    # value computed from the previous input.
+    P.integer = 3
+    assert items == [3]
+    assert i.rx.value == 3
+
+async def test_reactive_async_ref_not_synced_while_awaiting():
+    class Ref(param.Parameterized):
+        value = param.Integer(default=0, allow_refs=True)
+
+    irx = rx(1)
+    p = Ref(value=irx.rx.pipe(mul_slowly))
+    await async_wait_until(lambda: p.value == 2)
+    irx.rx.value = 2
+    assert p.value == 2
+    await async_wait_until(lambda: p.value == 4)
+
+async def test_reactive_async_superseded_updates_not_computed():
+    started, finished = [], []
+
+    async def mul(value):
+        started.append(value)
+        await asyncio.sleep(0.02)
+        finished.append(value)
+        return value*2
+
+    irx = rx(1)
+    async_rx = irx.rx.pipe(mul)
+    items = []
+    async_rx.rx.watch(items.append)
+    async_rx.rx.value
+    await async_wait_until(lambda: items == [2])
+    assert started == [1]
+
+    # The updates arrive without yielding to the event loop, so the tasks they
+    # schedule have not started by the time the next one supersedes them.
+    for value in (2, 3, 4, 5):
+        irx.rx.value = value
+
+    await async_wait_until(lambda: items == [2, 10])
+
+    # Only the final update was computed; the superseded coroutines were closed
+    # before their bodies began.
+    assert started == [1, 5]
+    assert finished == [1, 5]
+
+async def test_reactive_async_gen_superseded_updates_not_computed():
+    started = []
+
+    async def gen(value):
+        started.append(value)
+        yield value*2
+
+    irx = rx(1)
+    async_rx = irx.rx.pipe(gen)
+    async_rx.rx.watch()
+    async_rx.rx.value
+    await async_wait_until(lambda: async_rx.rx.value == 2)
+    assert started == [1]
+
+    for value in (2, 3, 4, 5):
+        irx.rx.value = value
+
+    await async_wait_until(lambda: async_rx.rx.value == 10)
+
+    # A superseded async generator is closed before it is first iterated, so
+    # its body never runs.
+    assert started == [1, 5]
+
+async def test_reactive_async_running_task_is_cancelled():
+    cancelled = []
+
+    async def mul(value):
+        try:
+            await asyncio.sleep(0.1)
+        except asyncio.CancelledError:
+            cancelled.append(value)
+            raise
+        return value*2
+
+    irx = rx(1)
+    async_rx = irx.rx.pipe(mul)
+    async_rx.rx.watch()
+    async_rx.rx.value
+
+    # Yield to the event loop so the body is actually suspended on its await;
+    # a task in that state cannot be skipped, it has to be cancelled.
+    await asyncio.sleep(0.01)
+    irx.rx.value = 2
+
+    await async_wait_until(lambda: async_rx.rx.value == 4)
+    assert cancelled == [1]
+
+@pytest.mark.parametrize('lazy', [False, True])
+async def test_async_shared_rx_superseded_updates_computed_once(lazy):
+    call_count = 0
+
+    class Model(param.Parameterized):
+        a = param.Number(1.0)
+
+    model = Model()
+
+    async def expensive_compute(a):
+        nonlocal call_count
+        call_count += 1
+        await asyncio.sleep(0.02)
+        return {"x": a + 1, "y": a * 2}
+
+    shared = rx(model.param.a, lazy=lazy).rx.pipe(expensive_compute)
+    x_rx = shared.rx.pipe(lambda d: d["x"])
+    y_rx = shared.rx.pipe(lambda d: d["y"])
+
+    x_rx.rx.value
+    y_rx.rx.value
+    await async_wait_until(lambda: call_count == 1)
+
+    for value in (2.0, 3.0, 4.0):
+        model.a = value
+
+    x_rx.rx.value
+    y_rx.rx.value
+    await async_wait_until(
+        lambda: x_rx.rx.value == 5 and y_rx.rx.value == 8
+    )
+
+    # The branches resolve through the shared node rather than recomputing, and
+    # the superseded updates are not computed at all.
+    assert call_count == 2
+
 @pytest.mark.parametrize('lazy', [False, True])
 def test_root_invalidation(lazy):
     arx = rx('a', lazy=lazy)