Avoid defining awaitable in PyActor.__init__

[mroeschke]

While running the Polars test suite with cudf_polars + rapidsmpf (with warnings as errors), I noticed this RuntimeError

________ test_grouped_agg_parametric[drop_nans-True-False-False-False] _________
[gw6] linux -- Python 3.12.13 /pyenv/versions/3.12.13/bin/python
Traceback (most recent call last):
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/_pytest/runner.py", line 341, in from_call
    result: TResult | None = func()
                             ^^^^^^
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/_pytest/runner.py", line 242, in <lambda>
    lambda: runtest_hook(item=item, **kwds), when=when, reraise=reraise
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/pluggy/_hooks.py", line 512, in __call__
    return self._hookexec(self.name, self._hookimpls.copy(), kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/pluggy/_manager.py", line 120, in _hookexec
    return self._inner_hookexec(hook_name, methods, kwargs, firstresult)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/pluggy/_callers.py", line 167, in _multicall
    raise exception
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/pluggy/_callers.py", line 139, in _multicall
    teardown.throw(exception)
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/_pytest/threadexception.py", line 92, in pytest_runtest_call
    yield from thread_exception_runtest_hook()
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/_pytest/threadexception.py", line 68, in thread_exception_runtest_hook
    yield
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/pluggy/_callers.py", line 152, in _multicall
    teardown.send(result)
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/_pytest/unraisableexception.py", line 95, in pytest_runtest_call
    yield from unraisable_exception_runtest_hook()
  File "/pyenv/versions/3.12.13/lib/python3.12/site-packages/_pytest/unraisableexception.py", line 85, in unraisable_exception_runtest_hook
    warnings.warn(pytest.PytestUnraisableExceptionWarning(msg))
pytest.PytestUnraisableExceptionWarning: Exception ignored in: <coroutine object dataframescan_node at 0x7d02cd317680>

Traceback (most recent call last):
  File "/pyenv/versions/3.12.13/lib/python3.12/warnings.py", line 555, in _warn_unawaited_coroutine
    warn(msg, category=RuntimeWarning, stacklevel=2, source=coro)
RuntimeWarning: coroutine 'dataframescan_node' was never awaited

Claude pointed me to PyActor creating an awaitable in __init__ that has a possibility of never being awaited and a solution to create this awaitable within __await__ instead.

I was able to reproduce a similar failure in test_unawaited_actor_closed_coroutines_no_warning without this fix that raises

(rapidsmpf-dev) /rapidsmpf$ pytest python/rapidsmpf/rapidsmpf/tests/streaming/test_define_actor.py
====================================================== test session starts ======================================================
platform linux -- Python 3.14.3, pytest-9.0.2, pluggy-1.6.0
rootdir: /rapidsmpf
configfile: pyproject.toml
plugins: anyio-4.13.0
collected 5 items                                                                                                               

python/rapidsmpf/rapidsmpf/tests/streaming/test_define_actor.py ....F                                                     [100%]

=========================================================== FAILURES ============================================================
_______________________________________ test_unawaited_actor_closed_coroutines_no_warning _______________________________________

coro = <coroutine object test_unawaited_actor_closed_coroutines_no_warning.<locals>.my_actor at 0x7e99b4fd5140>

    def _warn_unawaited_coroutine(coro):
        msg_lines = [
            f"coroutine '{coro.__qualname__}' was never awaited\n"
        ]
        if coro.cr_origin is not None:
            import linecache, traceback
            def extract():
                for filename, lineno, funcname in reversed(coro.cr_origin):
                    line = linecache.getline(filename, lineno)
                    yield (filename, lineno, funcname, line)
            msg_lines.append("Coroutine created at (most recent call last)\n")
            msg_lines += traceback.format_list(list(extract()))
        msg = "".join(msg_lines).rstrip("\n")
        # Passing source= here means that if the user happens to have tracemalloc
        # enabled and tracking where the coroutine was created, the warning will
        # contain that traceback. This does mean that if they have *both*
        # coroutine origin tracking *and* tracemalloc enabled, they'll get two
        # partially-redundant tracebacks. If we wanted to be clever we could
        # probably detect this case and avoid it, but for now we don't bother.
>       _wm.warn(
            msg, category=RuntimeWarning, stacklevel=2, source=coro
        )
E       RuntimeWarning: coroutine 'test_unawaited_actor_closed_coroutines_no_warning.<locals>.my_actor' was never awaited

../conda/envs/rapidsmpf-dev/lib/python3.14/_py_warnings.py:856: RuntimeWarning

[TomAugspurger]

This is fine, but I opened #942 to do it by default.

[TomAugspurger]

I think this looks good, but maybe @madsbk can take a look as well when he has a chance.

[madsbk]

Looks good, thanks @mroeschke

[TomAugspurger]

/merge