Settle an rx generator operation whose stream ends without yielding

[philippjfr]

An rx node whose operation is a generator stays "in flight" forever when the stream ends without yielding a value for the current inputs.

This PR adds ensures that when the loop completes without having produced a value it marks the node skipped so the value it computed from earlier inputs is not propagated as if it were current. That is the same outcome as an operation raising Skip: the pipeline responded to the current inputs by declining to produce a value, and the previous value stands

import asyncio
import param

async def gen(value):
    if value > 0:
        yield value * 2

async def main():
    number = param.rx(1)
    expr = number.rx.pipe(gen)
    expr.rx.watch()

    while expr.rx.value != 2:
        await asyncio.sleep(0.02)
    assert not expr.rx.awaiting

    number.rx.value = 0          # this stream yields nothing
    await asyncio.sleep(0.2)
    assert not expr.rx.awaiting  # fails on main: stuck True with no task running
    assert expr.rx.value == 2    # previous value stands, as with Skip

asyncio.run(main())

The stuck flag is observable through .rx.awaiting, and it also wedges anything built on the same generation machinery, so a consumer polling for an expression to settle waits forever instead of learning that no value is coming.

AI Disclosure

Tool & Model: Kilo Code + Claude Opus

Usage: Diagnosed the stuck generation from the observable awaiting behavior, wrote the fix and the tests, and confirmed the tests fail without the fix.

• I have tested all AI-generated content in my PR.
• I take responsibility for all AI-generated content in my PR.

Checklist

• Tests added and are passing
• Added documentation