Fix branching off an async rx node

[philippjfr]

A branch taken off an async node that has already settled never resolves. It reports
Undefined forever, silently, and no internal state flags the condition: the node is not dirty
and not awaiting, so nothing will ever schedule a resolution for it.

async def pair(v):
    await asyncio.sleep(0.01)
    return (v * 2, {"trace": v})

node = rx(1).rx.pipe(pair)
node.rx.value
await async_wait_until(lambda: node.rx.value == (2, {"trace": 1}))

first = node[0]
first.rx.value   # Undefined on main, forever

Branching a tuple- or dict-returning node into its components (node[0], node[1],
node.attr) is the normal way to fan a single computation out to several consumers, and for a
synchronous node it works and costs one compute. For an async node it depends entirely on when
the branch is created: before the node is driven it resolves, in flight it resolves, and after it
has settled it never resolves at all.

Cause

node[0] builds a two-node chain: a _clone(copy=True) mirror that carries the same async
operation with _shared pointing at the original, and the getitem node on top of it.

_clone(copy=True) passes _current=self._current, so __init__ computes
self._dirty = _current is None or _current is Undefined and gets False because the shared
node has a value. A few lines later the async branch of __init__ discards that value:

if operation and (iscoroutinefunction(operation['fn']) or inspect.isgeneratorfunction(operation['fn'])):
    self._trigger = Trigger(internal=True)
    self._current_ = Undefined     # _dirty is left False

The mirror is born clean while holding Undefined. _resolve therefore takes its else branch
and returns Undefined without scheduling anything, and the getitem node on top sees
obj is Undefined and skips. Generations stay at 0 == 0, so _awaiting is False, and only an
input change would ever mark the mirror dirty again.

The same applies to async generator nodes, which are cloned through the same branch of
__init__.

Fix

Two changes in param/reactive.py.

__init__ marks a node with an async operation dirty when it clears _current_. An async
operation has not resolved yet even when cloned from a node that has settled, so the node has to
be born dirty for the first read to resolve it.

_resolve's shared-input path then adopts the shared node's value synchronously when that node
has settled, instead of always scheduling a task to copy it one event loop iteration later:

shared = self._shared
value = shared.rx.value                  # triggers async resolve
if self._is_async and (shared._awaiting or shared._current_task is not None):
    self._lazy_resolve()                 # adopt it once the task is done
    raise Skip
self._current_ = value
self._skipped = shared._skipped

Adopting synchronously claims a generation (_resolve_generation/_finished_generation), which
supersedes a task an earlier read may have scheduled and keeps _awaiting from reporting a
resolution that is not in flight.

This also removes the second-read wart that motivated the workaround in downstream code: with
first, second = node[0], node[1], reading first to completion previously left second
returning Undefined on its first read and the value only on its next. The shared node has
settled by the time second is first read, so it now resolves immediately.

A branch created before the node is driven still returns Undefined on its first read; nothing
has been computed yet, and that is the ordinary demand-driven async pattern.

AI Disclosure

Developed with the assistance of Claude Opus 5 and manually tested.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 86.91%. Comparing base (83c4dca) to head (cc3667b).

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1178      +/-   ##
==========================================
+ Coverage   86.90%   86.91%   +0.01%     
==========================================
  Files           9        9              
  Lines        5398     5403       +5     
==========================================
+ Hits         4691     4696       +5     
  Misses        707      707              

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[hoxbro]

Will this always be an async resolve?

[philippjfr]

No

[hoxbro]

Then the comments should be updated right?

[philippjfr]

Depends on how you read it, it's only important for the the async case (so that's what it calls out) and is effectively a no-op for the synchronous case.

[hoxbro]

Previously it was behind a self._is_async which made the comment clear. The changes in this PR make the comment more obscure. I think we should add your comment here in the code.

[hoxbro]

As mentioned #1179 (comment). Having numbers only make it harder to read the test, than what it should be.