Draft test-doc for "out-of-band" `asyncio.Task`..

Since there's no way to activate `greenback`'s portal in such cases, we
should at least have a test verifying our very loud error about the
inability to support this usage..

pyproject.toml

@@ -2,12 +2,14 @@
 requires = ["hatchling"]
 build-backend = "hatchling.build"
 
+# ------ build-system ------
+
 [project]
 name = "tractor"
 version = "0.1.0a6dev0"
 description = 'structured concurrent `trio`-"actors"'
 authors = [{ name = "Tyler Goodlet", email = "goodboy_foss@protonmail.com" }]
-requires-python = "~=3.11"
+requires-python = ">= 3.11"
 readme = "docs/README.rst"
 license = "AGPL-3.0-or-later"
 keywords = [
@@ -30,32 +32,60 @@ classifiers = [
   "Topic :: System :: Distributed Computing",
 ]
 dependencies = [
+# trio runtime and friends
+  # (poetry) proper range specs,
+  # https://packaging.python.org/en/latest/discussions/install-requires-vs-requirements/#id5
+  # TODO, for 3.13 we must go go `0.27` which means we have to
+  # disable strict egs or port to handling them internally!
+  # trio='^0.27'
   "trio>=0.24,<0.25",
   "tricycle>=0.4.1,<0.5",
   "trio-typing>=0.10.0,<0.11",
+
   "wrapt>=1.16.0,<2",
   "colorlog>=6.8.2,<7",
+
+# built-in multi-actor `pdb` REPL
   "pdbp>=1.5.0,<2",
+
+# typed IPC msging
+# TODO, get back on release once 3.13 support is out!
   "msgspec",
 ]
 
+# ------ project ------
+
 [dependency-groups]
 dev = [
+  # test suite
+  # TODO: maybe some of these layout choices?
+  # https://docs.pytest.org/en/8.0.x/explanation/goodpractices.html#choosing-a-test-layout-import-rules
   "pytest>=8.2.0,<9",
   "pexpect>=4.9.0,<5",
+  # `tractor.devx` tooling
   "greenback>=1.2.1,<2",
   "stackscope>=0.2.2,<0.3",
+
+  # xonsh usage/integration (namely as @goodboy's sh of choice Bp)
+  "xonsh>=0.19.1",
   "xontrib-vox>=0.0.1,<0.0.2",
   "prompt-toolkit>=3.0.43,<4",
   "xonsh-vox-tabcomplete>=0.5,<0.6",
+  "pyperclip>=1.9.0",
 ]
 
+# TODO, distributed (multi-host) extensions
+# linux kernel networking
+# 'pyroute2
+
 [tool.hatch.build.targets.sdist]
 include = ["tractor"]
 
 [tool.hatch.build.targets.wheel]
 include = ["tractor"]
 
+# ------ dependency-groups ------
+
 [tool.towncrier]
 package = "tractor"
 filename = "NEWS.rst"
@@ -100,5 +130,9 @@ addopts = [
 ]
 log_cli = false
 
+# ------ tool.towncrier ------
+
 [tool.uv.sources]
 msgspec = { git = "https://github.com/jcrist/msgspec.git" }
+
+# ------ tool.uv.sources ------

ruff.toml

@@ -52,7 +52,7 @@ fixable = ["ALL"]
 unfixable = []
 
 # Allow unused variables when underscore-prefixed.
-dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
+# dummy-variable-rgx = "^(_+|(_+[a-zA-Z0-9_]*[a-zA-Z0-9]+?))$"
 
 [format]
 # Use single quotes in `ruff format`.

tests/conftest.py

@@ -150,6 +150,18 @@ def pytest_generate_tests(metafunc):
         metafunc.parametrize("start_method", [spawn_backend], scope='module')
 
 
+# TODO: a way to let test scripts (like from `examples/`)
+# guarantee they won't registry addr collide!
+# @pytest.fixture
+# def open_test_runtime(
+#     reg_addr: tuple,
+# ) -> AsyncContextManager:
+#     return partial(
+#         tractor.open_nursery,
+#         registry_addrs=[reg_addr],
+#     )
+
+
 def sig_prog(proc, sig):
     "Kill the actor-process with ``sig``."
     proc.send_signal(sig)

tests/devx/test_pause_from_non_trio.py

@@ -218,10 +218,9 @@ def expect_any_of(
         )
 
     return expected_patts
-    # yield child
 
 
-def test_pause_from_asyncio_task(
+def test_sync_pause_from_aio_task(
     spawn,
     ctlc: bool
     # ^TODO, fix for `asyncio`!!
@@ -327,3 +326,25 @@ def test_pause_from_asyncio_task(
 
     child.sendline('c')
     child.expect(EOF)
+
+
+def test_sync_pause_from_non_greenbacked_aio_task():
+    '''
+    Where the `breakpoint()` caller task is NOT spawned by
+    `tractor.to_asyncio` and thus never activates
+    a `greenback.ensure_portal()` beforehand, presumably bc the task
+    was started by some lib/dep as in often seen in the field.
+
+    Ensure sync pausing works when the pause is in,
+
+    - the root actor running in infected-mode?
+      |_ since we don't need any IPC to acquire the debug lock?
+      |_ is there some way to handle this like the non-main-thread case?
+
+    All other cases need to error out appropriately right?
+
+    - for any subactor we can't avoid needing the repl lock..
+      |_ is there a way to hook into `asyncio.ensure_future(obj)`?
+
+    '''
+    pass

tractor/devx/__init__.py

@@ -41,3 +41,38 @@ from .pformat import (
     pformat_caller_frame as pformat_caller_frame,
     pformat_boxed_tb as pformat_boxed_tb,
 )
+
+
+def _enable_readline_feats() -> str:
+    '''
+    Handle `readline` when compiled with `libedit` to avoid breaking
+    tab completion in `pdbp` (and its dep `tabcompleter`)
+    particularly since `uv` cpython distis are compiled this way..
+
+    See docs for deats,
+    https://docs.python.org/3/library/readline.html#module-readline
+
+    Originally discovered soln via SO answer,
+    https://stackoverflow.com/q/49287102
+
+    '''
+    import readline
+    if (
+        # 3.13+ attr
+        # https://docs.python.org/3/library/readline.html#readline.backend
+        (getattr(readline, 'backend', False) == 'libedit')
+        or
+        'libedit' in readline.__doc__
+    ):
+        readline.parse_and_bind("python:bind -v")
+        readline.parse_and_bind("python:bind ^I rl_complete")
+        return 'libedit'
+    else:
+        readline.parse_and_bind("tab: complete")
+        readline.parse_and_bind("set editing-mode vi")
+        readline.parse_and_bind("set keymap vi")
+        return 'readline'
+
+
+# TODO, move this to a new `.devx._pdbp` mod?
+_enable_readline_feats()

tractor/to_asyncio.py

@@ -34,7 +34,6 @@ from typing import (
 
 import tractor
 from tractor._exceptions import (
-    AsyncioCancelled,
     is_multi_cancelled,
 )
 from tractor._state import (
@@ -46,6 +45,11 @@ from tractor.log import (
     get_logger,
     StackLevelAdapter,
 )
+# TODO, wite the equiv of `trio.abc.Channel` but without attrs..
+# -[ ] `trionics.chan_types.ChanStruct` maybe?
+# from tractor.msg import (
+#     pretty_struct,
+# )
 from tractor.trionics._broadcast import (
     broadcast_receiver,
     BroadcastReceiver,
@@ -66,7 +70,12 @@ __all__ = [
 
 
 @dataclass
-class LinkedTaskChannel(trio.abc.Channel):
+class LinkedTaskChannel(
+    trio.abc.Channel,
+
+    # XXX LAME! meta-base conflict..
+    # pretty_struct.Struct,
+):
     '''
     A "linked task channel" which allows for two-way synchronized msg
     passing between a ``trio``-in-guest-mode task and an ``asyncio``
@@ -77,12 +86,14 @@ class LinkedTaskChannel(trio.abc.Channel):
     _from_aio: trio.MemoryReceiveChannel
     _to_trio: trio.MemorySendChannel
     _trio_cs: trio.CancelScope
+    _trio_task: trio.Task
     _aio_task_complete: trio.Event
 
     _trio_err: BaseException|None = None
     _trio_exited: bool = False
 
     # set after ``asyncio.create_task()``
+    # _aio_first: Any|None = None
     _aio_task: asyncio.Task|None = None
     _aio_err: BaseException|None = None
     _broadcaster: BroadcastReceiver|None = None
@@ -90,6 +101,25 @@ class LinkedTaskChannel(trio.abc.Channel):
     async def aclose(self) -> None:
         await self._from_aio.aclose()
 
+    def started(
+        self,
+        val: Any = None,
+    ) -> None:
+        self._aio_started_val = val
+        return self._to_trio.send_nowait(val)
+
+    # TODO, mk this side-agnostic?
+    #
+    # -[ ] add private meths for both sides and dynamically
+    #    determine which to use based on task-type read at calltime?
+    #    -[ ] `._recv_trio()`: receive to trio<-asyncio
+    #    -[ ] `._send_trio()`: send from trio->asyncio
+    #    -[ ] `._recv_aio()`: send from asyncio->trio
+    #    -[ ] `._send_aio()`: receive to asyncio<-trio
+    #
+    # -[ ] pass the instance to the aio side instead of the separate
+    #    per-side chan types?
+    #
     async def receive(self) -> Any:
         '''
         Receive a value from the paired `asyncio.Task` with
@@ -115,7 +145,16 @@ class LinkedTaskChannel(trio.abc.Channel):
             ):
                 raise err
 
-    async def wait_asyncio_complete(self) -> None:
+    async def send(self, item: Any) -> None:
+        '''
+        Send a value through to the asyncio task presuming
+        it defines a ``from_trio`` argument, if it does not
+        this method will raise an error.
+
+        '''
+        self._to_aio.put_nowait(item)
+
+    async def wait_aio_complete(self) -> None:
         await self._aio_task_complete.wait()
 
     def cancel_asyncio_task(
@@ -126,15 +165,6 @@ class LinkedTaskChannel(trio.abc.Channel):
             msg=msg,
         )
 
-    async def send(self, item: Any) -> None:
-        '''
-        Send a value through to the asyncio task presuming
-        it defines a ``from_trio`` argument, if it does not
-        this method will raise an error.
-
-        '''
-        self._to_aio.put_nowait(item)
-
     def closed(self) -> bool:
         return self._from_aio._closed  # type: ignore
 
@@ -218,7 +248,8 @@ def _run_asyncio_task(
 
     coro = func(**kwargs)
 
-    cancel_scope = trio.CancelScope()
+    trio_task: trio.Task = trio.lowlevel.current_task()
+    trio_cs = trio.CancelScope()
     aio_task_complete = trio.Event()
     aio_err: BaseException|None = None
 
@@ -226,7 +257,8 @@ def _run_asyncio_task(
         _to_aio=aio_q,  # asyncio.Queue
         _from_aio=from_aio,  # recv chan
         _to_trio=to_trio,  # send chan
-        _trio_cs=cancel_scope,
+        _trio_cs=trio_cs,
+        _trio_task=trio_task,
         _aio_task_complete=aio_task_complete,
     )
 
@@ -274,6 +306,9 @@ def _run_asyncio_task(
                 to_trio.send_nowait(result)
 
         finally:
+            # breakpoint()
+            # import pdbp; pdbp.set_trace()
+
             # if the task was spawned using `open_channel_from()`
             # then we close the channels on exit.
             if provide_channels:
@@ -281,7 +316,6 @@ def _run_asyncio_task(
                 # a ``trio.EndOfChannel`` to the trio (consumer) side.
                 to_trio.close()
 
-            # import pdbp; pdbp.set_trace()
             aio_task_complete.set()
             # await asyncio.sleep(0.1)
             log.info(
@@ -325,14 +359,17 @@ def _run_asyncio_task(
         )
         greenback.bestow_portal(task)
 
-    def cancel_trio(task: asyncio.Task) -> None:
+    def cancel_trio(
+        task: asyncio.Task,
+    ) -> None:
         '''
-        Cancel the calling `trio` task on error.
+        Cancel the parent `trio` task on any error raised by the
+        `asyncio` side.
 
         '''
         nonlocal chan
-        aio_err: BaseException|None = chan._aio_err
-        task_err: BaseException|None = None
+        relayed_aio_err: BaseException|None = chan._aio_err
+        aio_err: BaseException|None = None
 
         # only to avoid `asyncio` complaining about uncaptured
         # task exceptions
@@ -343,20 +380,20 @@ def _run_asyncio_task(
                 f'|_{res}\n'
             )
         except BaseException as _aio_err:
-            task_err: BaseException = _aio_err
-
+            aio_err: BaseException = _aio_err
             # read again AFTER the `asyncio` side errors in case
             # it was cancelled due to an error from `trio` (or
-            # some other out of band exc).
-            aio_err: BaseException|None = chan._aio_err
+            # some other out of band exc) and then set to something
+            # else?
+            relayed_aio_err: BaseException|None = chan._aio_err
 
             # always true right?
             assert (
-                type(_aio_err) is type(aio_err)
+                type(_aio_err) is type(relayed_aio_err)
             ), (
                 f'`asyncio`-side task errors mismatch?!?\n\n'
-                f'caught: {_aio_err}\n'
-                f'chan._aio_err: {aio_err}\n'
+                f'(caught) aio_err: {aio_err}\n'
+                f'chan._aio_err: {relayed_aio_err}\n'
             )
 
             msg: str = (
@@ -381,12 +418,13 @@ def _run_asyncio_task(
                     msg.format(etype_str='errored')
                 )
 
-
-        if aio_err is not None:
+        trio_err: BaseException|None = chan._trio_err
+        if (
+            relayed_aio_err
+            or
+            trio_err
+        ):
             # import pdbp; pdbp.set_trace()
-            # XXX: uhh is this true?
-            # assert task_err, f'Asyncio task {task.get_name()} discrepancy!?'
-
             # NOTE: currently mem chan closure may act as a form
             # of error relay (at least in the `asyncio.CancelledError`
             # case) since we have no way to directly trigger a `trio`
@@ -394,8 +432,6 @@ def _run_asyncio_task(
             # We might want to change this in the future though.
             from_aio.close()
 
-            if task_err is None:
-                assert aio_err
                 # wait, wut?
                 # aio_err.with_traceback(aio_err.__traceback__)
 
@@ -404,7 +440,7 @@ def _run_asyncio_task(
             # elif (
             #     type(aio_err) is CancelledError
             #     and  # trio was the cause?
-            #     cancel_scope.cancel_called
+            #     trio_cs.cancel_called
             # ):
             #     log.cancel(
             #         'infected task was cancelled by `trio`-side'
@@ -415,26 +451,83 @@ def _run_asyncio_task(
             # error in case the trio task is blocking on
             # a checkpoint.
             if (
-                not cancel_scope.cancelled_caught
+                not trio_cs.cancelled_caught
                 or
-                not cancel_scope.cancel_called
+                not trio_cs.cancel_called
             ):
                 # import pdbp; pdbp.set_trace()
-                cancel_scope.cancel()
+                trio_cs.cancel()
 
-            if task_err:
+            # maybe the `trio` task errored independent from the
+            # `asyncio` one and likely in between
+            # a guest-run-sched-tick.
+            #
+            # The obvious ex. is where one side errors during
+            # the current tick and then the other side immediately
+            # errors before its next checkpoint; i.e. the 2 errors
+            # are "independent".
+            #
+            # "Independent" here means in the sense that neither task
+            # was the explicit cause of the other side's exception
+            # according to our `tractor.to_asyncio` SC API's error
+            # relaying mechanism(s); the error pair is *possibly
+            # due-to* but **not necessarily** inter-related by some
+            # (subsys) state between the tasks,
+            #
+            # NOTE, also see the `test_trio_prestarted_task_bubbles`
+            # for reproducing detailed edge cases as per the above
+            # cases.
+            #
+            if (
+                not trio_cs.cancelled_caught
+                and
+                (trio_err := chan._trio_err)
+                and
+                type(trio_err) not in {
+                    trio.Cancelled,
+                }
+                and (
+                    aio_err
+                    and
+                    type(aio_err) not in {
+                        asyncio.CancelledError
+                    }
+                )
+            ):
+                eg = ExceptionGroup(
+                    'Both the `trio` and `asyncio` tasks errored independently!!\n',
+                    (trio_err, aio_err),
+                )
+                chan._trio_err = eg
+                chan._aio_err = eg
+                raise eg
+
+            elif aio_err:
                 # XXX raise any `asyncio` side error IFF it doesn't
                 # match the one we just caught from the task above!
                 # (that would indicate something weird/very-wrong
                 # going on?)
-                if aio_err is not task_err:
-                    # import pdbp; pdbp.set_trace()
-                    raise aio_err from task_err
+                if aio_err is not relayed_aio_err:
+                    raise aio_err from relayed_aio_err
+
+                raise aio_err
 
     task.add_done_callback(cancel_trio)
     return chan
 
 
+class AsyncioCancelled(Exception):
+    '''
+    Asyncio cancelled translation (non-base) error
+    for use with the ``to_asyncio`` module
+    to be raised in the ``trio`` side task
+
+    NOTE: this should NOT inherit from `asyncio.CancelledError` or
+    tests should break!
+
+    '''
+
+
 class TrioTaskExited(AsyncioCancelled):
     '''
     The `trio`-side task exited without explicitly cancelling the
@@ -483,13 +576,12 @@ async def translate_aio_errors(
 
         # import pdbp; pdbp.set_trace()  # lolevel-debug
 
-        # relay cancel through to called ``asyncio`` task
+        # relay cancel through to called `asyncio` task
         chan._aio_err = AsyncioCancelled(
             f'trio`-side cancelled the `asyncio`-side,\n'
             f'c)>\n'
             f'  |_{trio_task}\n\n'
 
-
             f'{trio_err!r}\n'
         )
 
@@ -546,6 +638,7 @@ async def translate_aio_errors(
             raise
 
     except BaseException as _trio_err:
+        # await tractor.pause(shield=True)
         trio_err = _trio_err
         log.exception(
             '`trio`-side task errored?'
@@ -619,11 +712,17 @@ async def translate_aio_errors(
                 # pump the other side's task? needed?
                 await trio.lowlevel.checkpoint()
 
+                # from tractor._state import is_root_process
+                # if is_root_process():
+                #     breakpoint()
+
                 if (
                     not chan._trio_err
                     and
                     (fut := aio_task._fut_waiter)
                 ):
+                    # await trio.lowlevel.checkpoint()
+                    # import pdbp; pdbp.set_trace()
                     fut.set_exception(
                         TrioTaskExited(
                             f'The peer `asyncio` task is still blocking/running?\n'
@@ -632,11 +731,6 @@ async def translate_aio_errors(
                         )
                     )
                 else:
-                    # from tractor._state import is_root_process
-                    # if is_root_process():
-                    #     breakpoint()
-                    #     import pdbp; pdbp.set_trace()
-
                     aio_taskc_warn: str = (
                         f'\n'
                         f'MANUALLY Cancelling `asyncio`-task: {aio_task.get_name()}!\n\n'
@@ -663,7 +757,7 @@ async def translate_aio_errors(
         # it erroed out up there!
         #
         if wait_on_aio_task:
-            # await chan.wait_asyncio_complete()
+            # await chan.wait_aio_complete()
             await chan._aio_task_complete.wait()
             log.info(
                 'asyncio-task is done and unblocked trio-side!\n'
@@ -771,11 +865,22 @@ async def open_channel_from(
             # sync to a "started()"-like first delivered value from the
             # ``asyncio`` task.
             try:
-                with chan._trio_cs:
+                with (cs := chan._trio_cs):
                     first = await chan.receive()
 
                     # deliver stream handle upward
                     yield first, chan
+            except trio.Cancelled as taskc:
+                # await tractor.pause(shield=True)  # ya it worx ;)
+                if cs.cancel_called:
+                    log.cancel(
+                        f'trio-side was manually cancelled by aio side\n'
+                        f'|_c>}}{cs!r}?\n'
+                    )
+                    # TODO, maybe a special `TrioCancelled`???
+
+                raise taskc
+
             finally:
                 chan._trio_exited = True
                 chan._to_trio.close()
@@ -957,15 +1062,15 @@ def run_as_asyncio_guest(
             #     force_reload=True,
             # )
 
-        def trio_done_callback(main_outcome):
+        def trio_done_callback(main_outcome: Outcome):
             log.runtime(
                 f'`trio` guest-run finishing with outcome\n'
                 f'>) {main_outcome}\n'
                 f'|_{trio_done_fute}\n'
             )
 
-            if isinstance(main_outcome, Error):
             # import pdbp; pdbp.set_trace()
+            if isinstance(main_outcome, Error):
                 error: BaseException = main_outcome.error
 
                 # show an dedicated `asyncio`-side tb from the error
@@ -1165,6 +1270,10 @@ def run_as_asyncio_guest(
                 )
                 raise AsyncioRuntimeTranslationError(message) from state_err
 
+        # XXX, should never get here ;)
+        # else:
+        #     import pdbp; pdbp.set_trace()
+
     # might as well if it's installed.
     try:
         import uvloop

uv.lock

@@ -1,5 +1,6 @@
 version = 1
-requires-python = ">=3.11, <4"
+revision = 1
+requires-python = ">=3.11"
 
 [[package]]
 name = "async-generator"
@@ -252,6 +253,12 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/8a/0b/9fcc47d19c48b59121088dd6da2488a49d5f72dacf8262e2790a1d2c7d15/pygments-2.19.1-py3-none-any.whl", hash = "sha256:9ea1544ad55cecf4b8242fab6dd35a93bbce657034b0611ee383099054ab6d8c", size = 1225293 },
 ]
 
+[[package]]
+name = "pyperclip"
+version = "1.9.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/30/23/2f0a3efc4d6a32f3b63cdff36cd398d9701d26cda58e3ab97ac79fb5e60d/pyperclip-1.9.0.tar.gz", hash = "sha256:b7de0142ddc81bfc5c7507eea19da920b92252b548b96186caf94a5e2527d310", size = 20961 }
+
 [[package]]
 name = "pyreadline3"
 version = "3.5.4"
@@ -334,8 +341,10 @@ dev = [
     { name = "greenback" },
     { name = "pexpect" },
     { name = "prompt-toolkit" },
+    { name = "pyperclip" },
     { name = "pytest" },
     { name = "stackscope" },
+    { name = "xonsh" },
     { name = "xonsh-vox-tabcomplete" },
     { name = "xontrib-vox" },
 ]
@@ -356,8 +365,10 @@ dev = [
     { name = "greenback", specifier = ">=1.2.1,<2" },
     { name = "pexpect", specifier = ">=4.9.0,<5" },
     { name = "prompt-toolkit", specifier = ">=3.0.43,<4" },
+    { name = "pyperclip", specifier = ">=1.9.0" },
     { name = "pytest", specifier = ">=8.2.0,<9" },
     { name = "stackscope", specifier = ">=0.2.2,<0.3" },
+    { name = "xonsh", specifier = ">=0.19.1" },
     { name = "xonsh-vox-tabcomplete", specifier = ">=0.5,<0.6" },
     { name = "xontrib-vox", specifier = ">=0.0.1,<0.0.2" },
 ]