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..

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

tests/test_infected_asyncio.py

@@ -5,6 +5,7 @@ The hipster way to force SC onto the stdlib's "async": 'infection mode'.
 import asyncio
 import builtins
 from contextlib import ExitStack
+# from functools import partial
 import itertools
 import importlib
 import os
@@ -108,7 +109,9 @@ async def asyncio_actor(
 
     except BaseException as err:
         if expect_err:
-            assert isinstance(err, error_type)
+            assert isinstance(err, error_type), (
+                f'{type(err)} is not {error_type}?'
+            )
 
         raise
 
@@ -180,8 +183,8 @@ def test_trio_cancels_aio(reg_addr):
         with trio.move_on_after(1):
             # cancel the nursery shortly after boot
 
-            async with tractor.open_nursery() as n:
-                await n.run_in_actor(
+            async with tractor.open_nursery() as tn:
+                await tn.run_in_actor(
                     asyncio_actor,
                     target='aio_sleep_forever',
                     expect_err='trio.Cancelled',
@@ -201,9 +204,12 @@ async def trio_ctx(
     # this will block until the ``asyncio`` task sends a "first"
     # message.
     with trio.fail_after(2):
+        try:
             async with (
-            trio.open_nursery() as n,
-
+                trio.open_nursery(
+                    # TODO, for new `trio` / py3.13
+                    # strict_exception_groups=False,
+                ) as tn,
                 tractor.to_asyncio.open_channel_from(
                     sleep_and_err,
                 ) as (first, chan),
@@ -212,12 +218,20 @@ async def trio_ctx(
                 assert first == 'start'
 
                 # spawn another asyncio task for the cuck of it.
-            n.start_soon(
+                tn.start_soon(
                     tractor.to_asyncio.run_task,
                     aio_sleep_forever,
                 )
                 await trio.sleep_forever()
 
+        # TODO, factor this into a `trionics.collapse()`?
+        except* BaseException as beg:
+            # await tractor.pause(shield=True)
+            if len(excs := beg.exceptions) == 1:
+                raise excs[0]
+            else:
+                raise
+
 
 @pytest.mark.parametrize(
     'parent_cancels',
@@ -235,7 +249,6 @@ def test_context_spawns_aio_task_that_errors(
 
     '''
     async def main():
-
         with trio.fail_after(2):
             async with tractor.open_nursery() as n:
                 p = await n.start_actor(
@@ -307,7 +320,9 @@ async def aio_cancel():
     await aio_sleep_forever()
 
 
-def test_aio_cancelled_from_aio_causes_trio_cancelled(reg_addr):
+def test_aio_cancelled_from_aio_causes_trio_cancelled(
+    reg_addr: tuple,
+):
     '''
     When the `asyncio.Task` cancels itself the `trio` side cshould
     also cancel and teardown and relay the cancellation cross-process
@@ -404,6 +419,7 @@ async def stream_from_aio(
             sequence=seq,
             expect_cancel=raise_err or exit_early,
             fail_early=aio_raise_err,
+
         ) as (first, chan):
 
             assert first is True
@@ -422,10 +438,15 @@ async def stream_from_aio(
                         if raise_err:
                             raise Exception
                         elif exit_early:
+                            print('`consume()` breaking early!\n')
                             break
 
+                print('returning from `consume()`..\n')
+
+            # run 2 tasks each pulling from
+            # the inter-task-channel with the 2nd
+            # using a fan-out `BroadcastReceiver`.
             if fan_out:
-                # start second task that get's the same stream value set.
                 async with (
 
                     # NOTE: this has to come first to avoid
@@ -435,11 +456,19 @@ async def stream_from_aio(
 
                     trio.open_nursery() as n,
                 ):
+                    # start 2nd task that get's broadcast the same
+                    # value set.
                     n.start_soon(consume, br)
                     await consume(chan)
 
             else:
                 await consume(chan)
+    except BaseException as err:
+        import logging
+        log = logging.getLogger()
+        log.exception('aio-subactor errored!\n')
+        raise err
+
     finally:
 
         if (
@@ -460,7 +489,8 @@ async def stream_from_aio(
             assert not fan_out
             assert pulled == expect[:51]
 
-        print('trio guest mode task completed!')
+        print('trio guest-mode task completed!')
+        assert chan._aio_task.done()
 
 
 @pytest.mark.parametrize(
@@ -500,19 +530,37 @@ def test_trio_error_cancels_intertask_chan(reg_addr):
     excinfo.value.boxed_type is Exception
 
 
-def test_trio_closes_early_and_channel_exits(reg_addr):
+def test_trio_closes_early_and_channel_exits(
+    reg_addr: tuple[str, int],
+):
+    '''
+    Check that if the `trio`-task "exits early" on `async for`ing the
+    inter-task-channel (via a `break`) we exit silently from the
+    `open_channel_from()` block and get a final `Return[None]` msg.
+
+    '''
     async def main():
-        async with tractor.open_nursery() as n:
+        with trio.fail_after(2):
+            async with tractor.open_nursery(
+                # debug_mode=True,
+                # enable_stack_on_sig=True,
+            ) as n:
                 portal = await n.run_in_actor(
                     stream_from_aio,
                     exit_early=True,
                     infect_asyncio=True,
                 )
                 # should raise RAE diectly
-            await portal.result()
+                print('waiting on final infected subactor result..')
+                res: None = await portal.wait_for_result()
+                assert res is None
+                print('infected subactor returned result: {res!r}\n')
 
     # should be a quiet exit on a simple channel exit
-    trio.run(main)
+    trio.run(
+        main,
+        # strict_exception_groups=False,
+    )
 
 
 def test_aio_errors_and_channel_propagates_and_closes(reg_addr):
@@ -536,11 +584,6 @@ def test_aio_errors_and_channel_propagates_and_closes(reg_addr):
     excinfo.value.boxed_type is Exception
 
 
-@tractor.context
-async def trio_to_aio_echo_server(
-    ctx: tractor.Context,
-):
-
 async def aio_echo_server(
     to_trio: trio.MemorySendChannel,
     from_trio: asyncio.Queue,
@@ -562,15 +605,19 @@ async def trio_to_aio_echo_server(
 
     print('exiting asyncio task')
 
+
+@tractor.context
+async def trio_to_aio_echo_server(
+    ctx: tractor.Context|None,
+):
     async with to_asyncio.open_channel_from(
         aio_echo_server,
     ) as (first, chan):
-
         assert first == 'start'
+
         await ctx.started(first)
 
         async with ctx.open_stream() as stream:
-
             async for msg in stream:
                 print(f'asyncio echoing {msg}')
                 await chan.send(msg)
@@ -649,7 +696,6 @@ def test_echoserver_detailed_mechanics(
         trio.run(main)
 
 
-
 @tractor.context
 async def manage_file(
     ctx: tractor.Context,

tests/test_root_infect_asyncio.py

@@ -0,0 +1,244 @@
+'''
+Special attention cases for using "infect `asyncio`" mode from a root
+actor; i.e. not using a std `trio.run()` bootstrap.
+
+'''
+import asyncio
+from functools import partial
+
+import pytest
+import trio
+import tractor
+from tractor import (
+    to_asyncio,
+)
+from tests.test_infected_asyncio import (
+    aio_echo_server,
+)
+
+
+@pytest.mark.parametrize(
+    'raise_error_mid_stream',
+    [
+        False,
+        Exception,
+        KeyboardInterrupt,
+    ],
+    ids='raise_error={}'.format,
+)
+def test_infected_root_actor(
+    raise_error_mid_stream: bool|Exception,
+
+    # conftest wide
+    loglevel: str,
+    debug_mode: bool,
+):
+    '''
+    Verify you can run the `tractor` runtime with `Actor.is_infected_aio() == True`
+    in the root actor.
+
+    '''
+    async def _trio_main():
+        with trio.fail_after(2):
+            first: str
+            chan: to_asyncio.LinkedTaskChannel
+            async with (
+                tractor.open_root_actor(
+                    debug_mode=debug_mode,
+                    loglevel=loglevel,
+                ),
+                to_asyncio.open_channel_from(
+                    aio_echo_server,
+                ) as (first, chan),
+            ):
+                assert first == 'start'
+
+                for i in range(1000):
+                    await chan.send(i)
+                    out = await chan.receive()
+                    assert out == i
+                    print(f'asyncio echoing {i}')
+
+                    if raise_error_mid_stream and i == 500:
+                        raise raise_error_mid_stream
+
+                    if out is None:
+                        try:
+                            out = await chan.receive()
+                        except trio.EndOfChannel:
+                            break
+                        else:
+                            raise RuntimeError(
+                                'aio channel never stopped?'
+                            )
+
+    if raise_error_mid_stream:
+        with pytest.raises(raise_error_mid_stream):
+            tractor.to_asyncio.run_as_asyncio_guest(
+                trio_main=_trio_main,
+            )
+    else:
+        tractor.to_asyncio.run_as_asyncio_guest(
+            trio_main=_trio_main,
+        )
+
+
+
+async def sync_and_err(
+    # just signature placeholders for compat with
+    # ``to_asyncio.open_channel_from()``
+    to_trio: trio.MemorySendChannel,
+    from_trio: asyncio.Queue,
+    ev: asyncio.Event,
+
+):
+    if to_trio:
+        to_trio.send_nowait('start')
+
+    await ev.wait()
+    raise RuntimeError('asyncio-side')
+
+
+@pytest.mark.parametrize(
+    'aio_err_trigger',
+    [
+        'before_start_point',
+        'after_trio_task_starts',
+        'after_start_point',
+    ],
+    ids='aio_err_triggered={}'.format
+)
+def test_trio_prestarted_task_bubbles(
+    aio_err_trigger: str,
+
+    # conftest wide
+    loglevel: str,
+    debug_mode: bool,
+):
+    async def pre_started_err(
+        raise_err: bool = False,
+        pre_sleep: float|None = None,
+        aio_trigger: asyncio.Event|None = None,
+        task_status=trio.TASK_STATUS_IGNORED,
+    ):
+        '''
+        Maybe pre-started error then sleep.
+
+        '''
+        if pre_sleep is not None:
+            print(f'Sleeping from trio for {pre_sleep!r}s !')
+            await trio.sleep(pre_sleep)
+
+        # signal aio-task to raise JUST AFTER this task
+        # starts but has not yet `.started()`
+        if aio_trigger:
+            print('Signalling aio-task to raise from `trio`!!')
+            aio_trigger.set()
+
+        if raise_err:
+            print('Raising from trio!')
+            raise TypeError('trio-side')
+
+        task_status.started()
+        await trio.sleep_forever()
+
+    async def _trio_main():
+        # with trio.fail_after(2):
+        with trio.fail_after(999):
+            first: str
+            chan: to_asyncio.LinkedTaskChannel
+            aio_ev = asyncio.Event()
+
+            async with (
+                tractor.open_root_actor(
+                    debug_mode=False,
+                    loglevel=loglevel,
+                ),
+            ):
+                # TODO, tests for this with 3.13 egs?
+                # from tractor.devx import open_crash_handler
+                # with open_crash_handler():
+                async with (
+                    # where we'll start a sub-task that errors BEFORE
+                    # calling `.started()` such that the error should
+                    # bubble before the guest run terminates!
+                    trio.open_nursery() as tn,
+
+                    # THEN start an infect task which should error just
+                    # after the trio-side's task does.
+                    to_asyncio.open_channel_from(
+                        partial(
+                            sync_and_err,
+                            ev=aio_ev,
+                        )
+                    ) as (first, chan),
+                ):
+
+                    for i in range(5):
+                        pre_sleep: float|None = None
+                        last_iter: bool = (i == 4)
+
+                        # TODO, missing cases?
+                        # -[ ] error as well on
+                        #    'after_start_point' case as well for
+                        #    another case?
+                        raise_err: bool = False
+
+                        if last_iter:
+                            raise_err: bool = True
+
+                            # trigger aio task to error on next loop
+                            # tick/checkpoint
+                            if aio_err_trigger == 'before_start_point':
+                                aio_ev.set()
+
+                            pre_sleep: float = 0
+
+                        await tn.start(
+                            pre_started_err,
+                            raise_err,
+                            pre_sleep,
+                            (aio_ev if (
+                                    aio_err_trigger == 'after_trio_task_starts'
+                                    and
+                                    last_iter
+                                ) else None
+                            ),
+                        )
+
+                        if (
+                            aio_err_trigger == 'after_start_point'
+                            and
+                            last_iter
+                        ):
+                            aio_ev.set()
+
+    with pytest.raises(
+        expected_exception=ExceptionGroup,
+    ) as excinfo:
+        tractor.to_asyncio.run_as_asyncio_guest(
+            trio_main=_trio_main,
+        )
+
+    eg = excinfo.value
+    rte_eg, rest_eg = eg.split(RuntimeError)
+
+    # ensure the trio-task's error bubbled despite the aio-side
+    # having (maybe) errored first.
+    if aio_err_trigger in (
+        'after_trio_task_starts',
+        'after_start_point',
+    ):
+        assert len(errs := rest_eg.exceptions) == 1
+        typerr = errs[0]
+        assert (
+            type(typerr) is TypeError
+            and
+            'trio-side' in typerr.args
+        )
+
+    # when aio errors BEFORE (last) trio task is scheduled, we should
+    # never see anythinb but the aio-side.
+    else:
+        assert len(rtes := rte_eg.exceptions) == 1
+        assert 'asyncio-side' in rtes[0].args[0]

tests/test_trioisms.py

@@ -3,6 +3,10 @@ Reminders for oddities in `trio` that we need to stay aware of and/or
 want to see changed.
 
 '''
+from contextlib import (
+    asynccontextmanager as acm,
+)
+
 import pytest
 import trio
 from trio import TaskStatus
@@ -80,3 +84,115 @@ def test_stashed_child_nursery(use_start_soon):
 
     with pytest.raises(NameError):
         trio.run(main)
+
+
+@pytest.mark.parametrize(
+    ('unmask_from_canc', 'canc_from_finally'),
+    [
+        (True, False),
+        (True, True),
+        pytest.param(False, True,
+                     marks=pytest.mark.xfail(reason="never raises!")
+        ),
+    ],
+    # TODO, ask ronny how to impl this .. XD
+    # ids='unmask_from_canc={0}, canc_from_finally={1}',#.format,
+)
+def test_acm_embedded_nursery_propagates_enter_err(
+    canc_from_finally: bool,
+    unmask_from_canc: bool,
+):
+    '''
+    Demo how a masking `trio.Cancelled` could be handled by unmasking from the
+    `.__context__` field when a user (by accident) re-raises from a `finally:`.
+
+    '''
+    import tractor
+
+    @acm
+    async def maybe_raise_from_masking_exc(
+        tn: trio.Nursery,
+        unmask_from: BaseException|None = trio.Cancelled
+
+        # TODO, maybe offer a collection?
+        # unmask_from: set[BaseException] = {
+        #     trio.Cancelled,
+        # },
+    ):
+        if not unmask_from:
+            yield
+            return
+
+        try:
+            yield
+        except* unmask_from as be_eg:
+
+            # TODO, if we offer `unmask_from: set`
+            # for masker_exc_type in unmask_from:
+
+            matches, rest = be_eg.split(unmask_from)
+            if not matches:
+                raise
+
+            for exc_match in be_eg.exceptions:
+                if (
+                    (exc_ctx := exc_match.__context__)
+                    and
+                    type(exc_ctx) not in {
+                        # trio.Cancelled,  # always by default?
+                        unmask_from,
+                    }
+                ):
+                    exc_ctx.add_note(
+                        f'\n'
+                        f'WARNING: the above error was masked by a {unmask_from!r} !?!\n'
+                        f'Are you always cancelling? Say from a `finally:` ?\n\n'
+
+                        f'{tn!r}'
+                    )
+                    raise exc_ctx from exc_match
+
+
+    @acm
+    async def wraps_tn_that_always_cancels():
+        async with (
+            trio.open_nursery() as tn,
+            maybe_raise_from_masking_exc(
+                tn=tn,
+                unmask_from=(
+                    trio.Cancelled
+                    if unmask_from_canc
+                    else None
+                ),
+            )
+        ):
+            try:
+                yield tn
+            finally:
+                if canc_from_finally:
+                    tn.cancel_scope.cancel()
+                    await trio.lowlevel.checkpoint()
+
+    async def _main():
+        with tractor.devx.open_crash_handler() as bxerr:
+            assert not bxerr.value
+
+            async with (
+                wraps_tn_that_always_cancels() as tn,
+            ):
+                assert not tn.cancel_scope.cancel_called
+                assert 0
+
+        assert (
+            (err := bxerr.value)
+            and
+            type(err) is AssertionError
+        )
+
+    with pytest.raises(ExceptionGroup) as excinfo:
+        trio.run(_main)
+
+    eg: ExceptionGroup = excinfo.value
+    assert_eg, rest_eg = eg.split(AssertionError)
+
+    assert len(assert_eg.exceptions) == 1

tractor/_exceptions.py

@@ -1146,19 +1146,51 @@ def unpack_error(
 
 
 def is_multi_cancelled(
-    exc: BaseException|BaseExceptionGroup
-) -> bool:
+    exc: BaseException|BaseExceptionGroup,
+
+    ignore_nested: set[BaseException] = set(),
+
+) -> bool|BaseExceptionGroup:
     '''
-    Predicate to determine if a possible ``BaseExceptionGroup`` contains
-    only ``trio.Cancelled`` sub-exceptions (and is likely the result of
-    cancelling a collection of subtasks.
+    Predicate to determine if an `BaseExceptionGroup` only contains
+    some (maybe nested) set of sub-grouped exceptions (like only
+    `trio.Cancelled`s which get swallowed silently by default) and is
+    thus the result of "gracefully cancelling" a collection of
+    sub-tasks (or other conc primitives) and receiving a "cancelled
+    ACK" from each after termination.
+
+    Docs:
+    ----
+    - https://docs.python.org/3/library/exceptions.html#exception-groups
+    - https://docs.python.org/3/library/exceptions.html#BaseExceptionGroup.subgroup
 
     '''
+
+    if (
+        not ignore_nested
+        or
+        trio.Cancelled in ignore_nested
+        # XXX always count-in `trio`'s native signal
+    ):
+        ignore_nested |= {trio.Cancelled}
+
     if isinstance(exc, BaseExceptionGroup):
-        return exc.subgroup(
-            lambda exc: isinstance(exc, trio.Cancelled)
-        ) is not None
+        matched_exc: BaseExceptionGroup|None = exc.subgroup(
+            tuple(ignore_nested),
 
+            # TODO, complain about why not allowed XD
+            # condition=tuple(ignore_nested),
+        )
+        if matched_exc is not None:
+            return matched_exc
+
+    # NOTE, IFF no excs types match (throughout the error-tree)
+    # -> return `False`, OW return the matched sub-eg.
+    #
+    # IOW, for the inverse of ^ for the purpose of
+    # maybe-enter-REPL--logic: "only debug when the err-tree contains
+    # at least one exc-type NOT in `ignore_nested`" ; i.e. the case where
+    # we fallthrough and return `False` here.
     return False
 
 

tractor/_root.py

@@ -95,6 +95,13 @@ async def open_root_actor(
 
     hide_tb: bool = True,
 
+    # XXX, proxied directly to `.devx._debug._maybe_enter_pm()`
+    # for REPL-entry logic.
+    debug_filter: Callable[
+        [BaseException|BaseExceptionGroup],
+        bool,
+    ] = lambda err: not is_multi_cancelled(err),
+
     # TODO, a way for actors to augment passing derived
     # read-only state to sublayers?
     # extra_rt_vars: dict|None = None,
@@ -334,6 +341,10 @@ async def open_root_actor(
             loglevel=loglevel,
             enable_modules=enable_modules,
         )
+        # XXX, in case the root actor runtime was actually run from
+        # `tractor.to_asyncio.run_as_asyncio_guest()` and NOt
+        # `.trio.run()`.
+        actor._infected_aio = _state._runtime_vars['_is_infected_aio']
 
     # Start up main task set via core actor-runtime nurseries.
     try:
@@ -375,6 +386,7 @@ async def open_root_actor(
                 Exception,
                 BaseExceptionGroup,
             ) as err:
+
                 # XXX NOTE XXX see equiv note inside
                 # `._runtime.Actor._stream_handler()` where in the
                 # non-root or root-that-opened-this-mahually case we
@@ -383,11 +395,15 @@ async def open_root_actor(
                 entered: bool = await _debug._maybe_enter_pm(
                     err,
                     api_frame=inspect.currentframe(),
+                    debug_filter=debug_filter,
                 )
+
                 if (
                     not entered
                     and
-                    not is_multi_cancelled(err)
+                    not is_multi_cancelled(
+                        err,
+                    )
                 ):
                     logger.exception('Root actor crashed\n')
 

tractor/devx/_debug.py

@@ -75,6 +75,7 @@ from tractor import _state
 from tractor._exceptions import (
     InternalError,
     NoRuntime,
+    is_multi_cancelled,
 )
 from tractor._state import (
     current_actor,
@@ -316,6 +317,7 @@ class Lock:
         we_released: bool = False
         ctx_in_debug: Context|None = cls.ctx_in_debug
         repl_task: Task|Thread|None = DebugStatus.repl_task
+        message: str = ''
 
         try:
             if not DebugStatus.is_main_trio_thread():
@@ -443,7 +445,10 @@ class Lock:
                         f'|_{repl_task}\n'
                     )
 
+            if message:
                 log.devx(message)
+            else:
+                import pdbp; pdbp.set_trace()
 
         return we_released
 
@@ -1743,7 +1748,7 @@ async def _pause(
     ] = trio.TASK_STATUS_IGNORED,
     **debug_func_kwargs,
 
-) -> tuple[PdbREPL, Task]|None:
+) -> tuple[Task, PdbREPL]|None:
     '''
     Inner impl for `pause()` to avoid the `trio.CancelScope.__exit__()`
     stack frame when not shielded (since apparently i can't figure out
@@ -1929,7 +1934,7 @@ async def _pause(
                 )
                 with trio.CancelScope(shield=shield):
                     await trio.lowlevel.checkpoint()
-                return repl, task
+                return (repl, task)
 
             # elif repl_task:
             #     log.warning(
@@ -2530,26 +2535,17 @@ def pause_from_sync(
             f'{actor.uid} task called `tractor.pause_from_sync()`\n'
         )
 
-        # TODO: once supported, remove this AND the one
-        # inside `._pause()`!
-        # outstanding impl fixes:
-        # -[ ] need to make `.shield_sigint()` below work here!
-        # -[ ] how to handle `asyncio`'s new SIGINT-handler
-        #     injection?
-        # -[ ] should `breakpoint()` work and what does it normally
-        #     do in `asyncio` ctxs?
-        # if actor.is_infected_aio():
-        #     raise RuntimeError(
-        #         '`tractor.pause[_from_sync]()` not yet supported '
-        #         'for infected `asyncio` mode!'
-        #     )
-
         repl: PdbREPL = mk_pdb()
 
         # message += f'-> created local REPL {repl}\n'
         is_trio_thread: bool = DebugStatus.is_main_trio_thread()
         is_root: bool = is_root_process()
-        is_aio: bool = actor.is_infected_aio()
+        is_infected_aio: bool = actor.is_infected_aio()
+        thread: Thread = threading.current_thread()
+
+        asyncio_task: asyncio.Task|None = None
+        if is_infected_aio:
+            asyncio_task = asyncio.current_task()
 
         # TODO: we could also check for a non-`.to_thread` context
         # using `trio.from_thread.check_cancelled()` (says
@@ -2565,20 +2561,14 @@ def pause_from_sync(
         if (
             not is_trio_thread
             and
-            not is_aio  # see below for this usage
+            not asyncio_task
         ):
             # TODO: `threading.Lock()` this so we don't get races in
             # multi-thr cases where they're acquiring/releasing the
             # REPL and setting request/`Lock` state, etc..
-            thread: threading.Thread = threading.current_thread()
-            repl_owner = thread
+            repl_owner: Thread = thread
 
             # TODO: make root-actor bg thread usage work!
-            if (
-                is_root
-                # or
-                # is_aio
-            ):
             if is_root:
                 message += (
                     f'-> called from a root-actor bg {thread}\n'
@@ -2637,13 +2627,32 @@ def pause_from_sync(
                 DebugStatus.shield_sigint()
                 assert bg_task is not DebugStatus.repl_task
 
+        # TODO: once supported, remove this AND the one
+        # inside `._pause()`!
+        # outstanding impl fixes:
+        # -[ ] need to make `.shield_sigint()` below work here!
+        # -[ ] how to handle `asyncio`'s new SIGINT-handler
+        #     injection?
+        # -[ ] should `breakpoint()` work and what does it normally
+        #     do in `asyncio` ctxs?
+        # if actor.is_infected_aio():
+        #     raise RuntimeError(
+        #         '`tractor.pause[_from_sync]()` not yet supported '
+        #         'for infected `asyncio` mode!'
+        #     )
         elif (
             not is_trio_thread
             and
-            is_aio
+            is_infected_aio  # as in, the special actor-runtime mode
+            # ^NOTE XXX, that doesn't mean the caller is necessarily
+            # an `asyncio.Task` just that `trio` has been embedded on
+            # the `asyncio` event loop!
+            and
+            asyncio_task  # transitive caller is an actual `asyncio.Task`
         ):
             greenback: ModuleType = maybe_import_greenback()
-            repl_owner: Task = asyncio.current_task()
+
+            if greenback.has_portal():
                 DebugStatus.shield_sigint()
                 fute: asyncio.Future = run_trio_task_in_future(
                     partial(
@@ -2660,11 +2669,53 @@ def pause_from_sync(
                         **_pause_kwargs
                     )
                 )
-
-            # TODO: for async version -> `.pause_from_aio()`?
-            # bg_task, _ = await fute
+                repl_owner = asyncio_task
                 bg_task, _ = greenback.await_(fute)
-            bg_task: asyncio.Task = asyncio.current_task()
+                # TODO: ASYNC version -> `.pause_from_aio()`?
+                # bg_task, _ = await fute
+
+            # handle the case where an `asyncio` task has been
+            # spawned WITHOUT enabling a `greenback` portal..
+            # => can often happen in 3rd party libs.
+            else:
+                bg_task = repl_owner
+
+                # TODO, ostensibly we can just acquire the
+                # debug lock directly presuming we're the
+                # root actor running in infected asyncio
+                # mode?
+                #
+                # TODO, this would be a special case where
+                # a `_pause_from_root()` would come in very
+                # handy!
+                # if is_root:
+                #     import pdbp; pdbp.set_trace()
+                #     log.warning(
+                #         'Allowing `asyncio` task to acquire debug-lock in root-actor..\n'
+                #         'This is not fully implemented yet; there may be teardown hangs!\n\n'
+                #     )
+                # else:
+
+                # simply unsupported, since there exists no hack (i
+                # can think of) to workaround this in a subactor
+                # which needs to lock the root's REPL ow we're sure
+                # to get prompt stdstreams clobbering..
+                cf_repr: str = ''
+                if api_frame:
+                    caller_frame: FrameType = api_frame.f_back
+                    cf_repr: str = f'caller_frame: {caller_frame!r}\n'
+
+                raise RuntimeError(
+                    f"CAN'T USE `greenback._await()` without a portal !?\n\n"
+                    f'Likely this task was NOT spawned via the `tractor.to_asyncio` API..\n'
+                    f'{asyncio_task}\n'
+                    f'{cf_repr}\n'
+
+                    f'Prolly the task was started out-of-band (from some lib?)\n'
+                    f'AND one of the below was never called ??\n'
+                    f'- greenback.ensure_portal()\n'
+                    f'- greenback.bestow_portal(<task>)\n'
+                )
 
         else:  # we are presumably the `trio.run()` + main thread
             # raises on not-found by default
@@ -2915,8 +2966,14 @@ async def _maybe_enter_pm(
     tb: TracebackType|None = None,
     api_frame: FrameType|None = None,
     hide_tb: bool = False,
+
+    # only enter debugger REPL when returns `True`
+    debug_filter: Callable[
+        [BaseException|BaseExceptionGroup],
+        bool,
+    ] = lambda err: not is_multi_cancelled(err),
+
 ):
-    from tractor._exceptions import is_multi_cancelled
     if (
         debug_mode()
 
@@ -2933,7 +2990,8 @@ async def _maybe_enter_pm(
 
         # Really we just want to mostly avoid catching KBIs here so there
         # might be a simpler check we can do?
-        and not is_multi_cancelled(err)
+        and
+        debug_filter(err)
     ):
         api_frame: FrameType = api_frame or inspect.currentframe()
         tb: TracebackType = tb or sys.exc_info()[2]
@@ -3114,7 +3172,7 @@ async def maybe_wait_for_debugger(
 @cm
 def open_crash_handler(
     catch: set[BaseException] = {
-        Exception,
+        # Exception,
         BaseException,
     },
     ignore: set[BaseException] = {
@@ -3135,14 +3193,30 @@ def open_crash_handler(
     '''
     __tracebackhide__: bool = tb_hide
 
+    class BoxedMaybeException(Struct):
+        value: BaseException|None = None
+
+    # TODO, yield a `outcome.Error`-like boxed type?
+    # -[~] use `outcome.Value/Error` X-> frozen!
+    # -[x] write our own..?
+    # -[ ] consider just wtv is used by `pytest.raises()`?
+    #
+    boxed_maybe_exc = BoxedMaybeException()
     err: BaseException
     try:
-        yield
+        yield boxed_maybe_exc
     except tuple(catch) as err:
-        if type(err) not in ignore:
-
-            # use our re-impl-ed version
+        boxed_maybe_exc.value = err
+        if (
+            type(err) not in ignore
+            and
+            not is_multi_cancelled(
+                err,
+                ignore_nested=ignore
+            )
+        ):
             try:
+                # use our re-impl-ed version
                 _post_mortem(
                     repl=mk_pdb(),
                     tb=sys.exc_info()[2],
@@ -3150,13 +3224,13 @@ def open_crash_handler(
                 )
             except bdb.BdbQuit:
                 __tracebackhide__: bool = False
-                raise
+                raise err
 
             # XXX NOTE, `pdbp`'s version seems to lose the up-stack
             # tb-info?
             # pdbp.xpm()
 
-        raise
+        raise err
 
 
 @cm

tractor/devx/pformat.py

@@ -92,7 +92,7 @@ def pformat_boxed_tb(
         f' ------ {boxer_header} ------\n'
         f'{tb_body}'
         f' ------ {boxer_header}- ------\n'
-        f'_|\n'
+        f'_|'
     )
     tb_box_indent: str = (
         tb_box_indent

tractor/log.py

@@ -258,20 +258,28 @@ class ActorContextInfo(Mapping):
 
 
 def get_logger(
-
     name: str|None = None,
     _root_name: str = _proj_name,
 
+    logger: Logger|None = None,
+
+    # TODO, using `.config.dictConfig()` api?
+    # -[ ] SO answer with docs links
+    #  |_https://stackoverflow.com/questions/7507825/where-is-a-complete-example-of-logging-config-dictconfig
+    #  |_https://docs.python.org/3/library/logging.config.html#configuration-dictionary-schema
+    subsys_spec: str|None = None,
+
 ) -> StackLevelAdapter:
     '''Return the package log or a sub-logger for ``name`` if provided.
 
     '''
     log: Logger
-    log = rlog = logging.getLogger(_root_name)
+    log = rlog = logger or logging.getLogger(_root_name)
 
     if (
         name
-        and name != _proj_name
+        and
+        name != _proj_name
     ):
 
         # NOTE: for handling for modules that use ``get_logger(__name__)``
@@ -306,7 +314,10 @@ def get_logger(
 
     # add our actor-task aware adapter which will dynamically look up
     # the actor and task names at each log emit
-    logger = StackLevelAdapter(log, ActorContextInfo())
+    logger = StackLevelAdapter(
+        log,
+        ActorContextInfo(),
+    )
 
     # additional levels
     for name, val in CUSTOM_LEVELS.items():
@@ -320,14 +331,24 @@ def get_logger(
 
 def get_console_log(
     level: str|None = None,
+    logger: Logger|None = None,
     **kwargs,
-) -> LoggerAdapter:
-    '''Get the package logger and enable a handler which writes to stderr.
 
-    Yeah yeah, i know we can use ``DictConfig``. You do it.
+) -> LoggerAdapter:
     '''
-    log = get_logger(**kwargs)  # our root logger
-    logger = log.logger
+    Get a `tractor`-style logging instance: a `Logger` wrapped in
+    a `StackLevelAdapter` which injects various concurrency-primitive
+    (process, thread, task) fields and enables a `StreamHandler` that
+    writes on stderr using `colorlog` formatting.
+
+    Yeah yeah, i know we can use `logging.config.dictConfig()`. You do it.
+
+    '''
+    log = get_logger(
+        logger=logger,
+        **kwargs
+    )  # set a root logger
+    logger: Logger = log.logger
 
     if not level:
         return log
@@ -346,9 +367,13 @@ def get_console_log(
             None,
         )
     ):
+        fmt = LOG_FORMAT
+        # if logger:
+        #     fmt = None
+
         handler = StreamHandler()
         formatter = colorlog.ColoredFormatter(
-            LOG_FORMAT,
+            fmt=fmt,
             datefmt=DATE_FORMAT,
             log_colors=STD_PALETTE,
             secondary_log_colors=BOLD_PALETTE,
@@ -365,7 +390,7 @@ def get_loglevel() -> str:
 
 
 # global module logger for tractor itself
-log = get_logger('tractor')
+log: StackLevelAdapter = get_logger('tractor')
 
 
 def at_least_level(

tractor/to_asyncio.py

@@ -33,12 +33,23 @@ from typing import (
 )
 
 import tractor
-from tractor._exceptions import AsyncioCancelled
+from tractor._exceptions import (
+    is_multi_cancelled,
+)
 from tractor._state import (
     debug_mode,
+    _runtime_vars,
 )
 from tractor.devx import _debug
-from tractor.log import get_logger
+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,
@@ -49,7 +60,7 @@ from outcome import (
     Outcome,
 )
 
-log = get_logger(__name__)
+log: StackLevelAdapter = get_logger(__name__)
 
 
 __all__ = [
@@ -59,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``
@@ -69,12 +85,15 @@ class LinkedTaskChannel(trio.abc.Channel):
     _to_aio: asyncio.Queue
     _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
@@ -82,7 +101,41 @@ 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
+        exception/cancel handling to teardown both sides on any
+        unexpected error.
+
+        '''
+        try:
+            # TODO: do we need this to guarantee asyncio code get's
+            # cancelled in the case where the trio side somehow creates
+            # a state where the asyncio cycle-task isn't getting the
+            # cancel request sent by (in theory) the last checkpoint
+            # cycle on the trio side?
+            # await trio.lowlevel.checkpoint()
+            return await self._from_aio.receive()
+        except BaseException as err:
             async with translate_aio_errors(
                 self,
 
@@ -90,21 +143,7 @@ class LinkedTaskChannel(trio.abc.Channel):
                 # being procesed.
                 # wait_on_aio_task=False,
             ):
-
-            # TODO: do we need this to guarantee asyncio code get's
-            # cancelled in the case where the trio side somehow creates
-            # a state where the asyncio cycle-task isn't getting the
-            # cancel request sent by (in theory) the last checkpoint
-            # cycle on the trio side?
-            # await trio.lowlevel.checkpoint()
-
-            return await self._from_aio.receive()
-
-    async def wait_asyncio_complete(self) -> None:
-        await self._aio_task_complete.wait()
-
-    # def cancel_asyncio_task(self) -> None:
-    #     self._aio_task.cancel()
+                raise err
 
     async def send(self, item: Any) -> None:
         '''
@@ -115,6 +154,17 @@ class LinkedTaskChannel(trio.abc.Channel):
         '''
         self._to_aio.put_nowait(item)
 
+    async def wait_aio_complete(self) -> None:
+        await self._aio_task_complete.wait()
+
+    def cancel_asyncio_task(
+        self,
+        msg: str = '',
+    ) -> None:
+        self._aio_task.cancel(
+            msg=msg,
+        )
+
     def closed(self) -> bool:
         return self._from_aio._closed  # type: ignore
 
@@ -154,7 +204,6 @@ class LinkedTaskChannel(trio.abc.Channel):
 
 
 def _run_asyncio_task(
-
     func: Callable,
     *,
     qsize: int = 1,
@@ -164,8 +213,9 @@ def _run_asyncio_task(
 
 ) -> LinkedTaskChannel:
     '''
-    Run an ``asyncio`` async function or generator in a task, return
-    or stream the result back to the caller `trio.lowleve.Task`.
+    Run an `asyncio`-compat async function or generator in a task,
+    return or stream the result back to the caller
+    `trio.lowleve.Task`.
 
     '''
     __tracebackhide__: bool = hide_tb
@@ -198,28 +248,30 @@ 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
 
     chan = LinkedTaskChannel(
-        aio_q,  # asyncio.Queue
-        from_aio,  # recv chan
-        to_trio,  # send chan
-
-        cancel_scope,
-        aio_task_complete,
+        _to_aio=aio_q,  # asyncio.Queue
+        _from_aio=from_aio,  # recv chan
+        _to_trio=to_trio,  # send chan
+        _trio_cs=trio_cs,
+        _trio_task=trio_task,
+        _aio_task_complete=aio_task_complete,
     )
 
     async def wait_on_coro_final_result(
-
         to_trio: trio.MemorySendChannel,
         coro: Awaitable,
         aio_task_complete: trio.Event,
 
     ) -> None:
         '''
-        Await ``coro`` and relay result back to ``trio``.
+        Await `coro` and relay result back to `trio`.
+
+        This can only be run as an `asyncio.Task`!
 
         '''
         nonlocal aio_err
@@ -242,8 +294,10 @@ def _run_asyncio_task(
 
         else:
             if (
-                result != orig and
-                aio_err is None and
+                result != orig
+                and
+                aio_err is None
+                and
 
                 # in the `open_channel_from()` case we don't
                 # relay through the "return value".
@@ -252,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:
@@ -260,11 +317,19 @@ def _run_asyncio_task(
                 to_trio.close()
 
             aio_task_complete.set()
-            log.runtime(f'`asyncio` task: {task.get_name()} is complete')
+            # await asyncio.sleep(0.1)
+            log.info(
+                f'`asyncio` task terminated\n'
+                f'x)>\n'
+                f'  |_{task}\n'
+            )
 
     # start the asyncio task we submitted from trio
     if not inspect.isawaitable(coro):
-        raise TypeError(f"No support for invoking {coro}")
+        raise TypeError(
+            f'Pass the async-fn NOT a coroutine\n'
+            f'{coro!r}'
+        )
 
     task: asyncio.Task = asyncio.create_task(
         wait_on_coro_final_result(
@@ -288,28 +353,55 @@ def _run_asyncio_task(
             raise_not_found=False,
         ))
     ):
+        log.info(
+            f'Bestowing `greenback` portal for `asyncio`-task\n'
+            f'{task}\n'
+        )
         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
         try:
             res: Any = task.result()
-        except BaseException as terr:
-            task_err: BaseException = terr
+            log.info(
+                '`trio` received final result from {task}\n'
+                f'|_{res}\n'
+            )
+        except BaseException as _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) and then set to something
+            # else?
+            relayed_aio_err: BaseException|None = chan._aio_err
+
+            # always true right?
+            assert (
+                type(_aio_err) is type(relayed_aio_err)
+            ), (
+                f'`asyncio`-side task errors mismatch?!?\n\n'
+                f'(caught) aio_err: {aio_err}\n'
+                f'chan._aio_err: {relayed_aio_err}\n'
+            )
 
             msg: str = (
-                'Infected `asyncio` task {etype_str}\n'
+                '`trio`-side reports that the `asyncio`-side '
+                '{etype_str}\n'
+                # ^NOTE filled in below
             )
-            if isinstance(terr, CancelledError):
+            if isinstance(_aio_err, CancelledError):
                 msg += (
                     f'c)>\n'
                     f' |_{task}\n'
@@ -326,75 +418,143 @@ def _run_asyncio_task(
                     msg.format(etype_str='errored')
                 )
 
-            assert type(terr) is type(aio_err), (
-                '`asyncio` task error mismatch?!?'
-            )
-
-        if aio_err is not None:
-            # XXX: uhh is this true?
-            # assert task_err, f'Asyncio task {task.get_name()} discrepancy!?'
-
+        trio_err: BaseException|None = chan._trio_err
+        if (
+            relayed_aio_err
+            or
+            trio_err
+        ):
+            # import pdbp; pdbp.set_trace()
             # 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``
+            # of error relay (at least in the `asyncio.CancelledError`
+            # case) since we have no way to directly trigger a `trio`
             # task error without creating a nursery to throw one.
             # 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__)
 
             # TODO: show when cancellation originated
-            # from each side more pedantically?
+            # from each side more pedantically in log-msg?
             # 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'
             #     )
             #     raise aio_err from task_err
 
-            # XXX: if not already, alway cancel the scope
-            # on a task error in case the trio task is blocking on
+            # XXX: if not already, alway cancel the scope on a task
+            # error in case the trio task is blocking on
             # a checkpoint.
-            cancel_scope.cancel()
-
             if (
-                task_err
-                and
-                aio_err is not task_err
+                not trio_cs.cancelled_caught
+                or
+                not trio_cs.cancel_called
             ):
-                raise aio_err from task_err
+                # import pdbp; pdbp.set_trace()
+                trio_cs.cancel()
 
-            # raise any `asyncio` side error.
-            raise aio_err
-
-        log.info(
-            '`trio` received final result from {task}\n'
-            f'|_{res}\n'
+            # 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
+                    }
                 )
-        # TODO: do we need this?
-        # if task_err:
-        #     cancel_scope.cancel()
-        #     raise task_err
+            ):
+                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 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
+    `asyncio.Task` peer.
+
+    This is very similar to how `trio.ClosedResource` acts as
+    a "clean shutdown" signal to the consumer side of a mem-chan,
+
+    https://trio.readthedocs.io/en/stable/reference-core.html#clean-shutdown-with-channels
+
+    '''
+
+
 @acm
 async def translate_aio_errors(
-
     chan: LinkedTaskChannel,
     wait_on_aio_task: bool = False,
+    cancel_aio_task_on_trio_exit: bool = True,
 
 ) -> AsyncIterator[None]:
     '''
-    Error handling context around ``asyncio`` task spawns which
+    An error handling to cross-loop propagation context around
+    `asyncio.Task` spawns via one of this module's APIs:
+
+    - `open_channel_from()`
+    - `run_task()`
+
     appropriately translates errors and cancels into ``trio`` land.
 
     '''
@@ -402,88 +562,248 @@ async def translate_aio_errors(
 
     aio_err: BaseException|None = None
 
-    # TODO: make thisi a channel method?
-    def maybe_raise_aio_err(
-        err: Exception|None = None
-    ) -> None:
-        aio_err = chan._aio_err
-        if (
-            aio_err is not None
-            and
-            # not isinstance(aio_err, CancelledError)
-            type(aio_err) != CancelledError
-        ):
-            # always raise from any captured asyncio error
-            if err:
-                raise aio_err from err
-            else:
-                raise aio_err
-
-    task = chan._aio_task
-    assert task
+    aio_task: asyncio.Task = chan._aio_task
+    assert aio_task
+    trio_err: BaseException|None = None
     try:
-        yield
+        yield  # back to one of the cross-loop apis
+    except trio.Cancelled as taskc:
+        trio_err = taskc
 
-    except (
-        trio.Cancelled,
-    ):
-        # relay cancel through to called ``asyncio`` task
+        # should NEVER be the case that `trio` is cancel-handling
+        # BEFORE the other side's task-ref was set!?
         assert chan._aio_task
-        chan._aio_task.cancel(
-            msg=f'the `trio` caller task was cancelled: {trio_task.name}'
-        )
-        raise
 
+        # import pdbp; pdbp.set_trace()  # lolevel-debug
+
+        # 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'
+        )
+
+        # XXX NOTE XXX seems like we can get all sorts of unreliable
+        # behaviour from `asyncio` under various cancellation
+        # conditions (like SIGINT/kbi) when this is used..
+        # SO FOR NOW, try to avoid it at most costs!
+        #
+        # aio_task.cancel(
+        #     msg=f'the `trio` parent task was cancelled: {trio_task.name}'
+        # )
+        # raise
+
+    # NOTE ALSO SEE the matching note in the `cancel_trio()` asyncio
+    # task-done-callback.
     except (
-        # NOTE: see the note in the ``cancel_trio()`` asyncio task
-        # termination callback
         trio.ClosedResourceError,
         # trio.BrokenResourceError,
-    ):
+    ) as cre:
+        trio_err = cre
         aio_err = chan._aio_err
+        # import pdbp; pdbp.set_trace()
+
+        # XXX if an underlying `asyncio.CancelledError` triggered
+        # this channel close, raise our (non-`BaseException`) wrapper
+        # exception (`AsyncioCancelled`) from that source error.
         if (
-            task.cancelled()
+            # aio-side is cancelled?
+            aio_task.cancelled()  # not set until it terminates??
             and
             type(aio_err) is CancelledError
+
+            # TODO, if we want suppression of the
+            # silent-exit-by-`trio` case?
+            # -[ ] the parent task can also just catch it though?
+            # -[ ] OR, offer a `signal_aio_side_on_exit=True` ??
+            #
+            # or
+            # aio_err is None
+            # and
+            # chan._trio_exited
+
         ):
-            # if an underlying `asyncio.CancelledError` triggered this
-            # channel close, raise our (non-``BaseException``) wrapper
-            # error: ``AsyncioCancelled`` from that source error.
             raise AsyncioCancelled(
-                f'Task cancelled\n'
-                f'|_{task}\n'
+                f'asyncio`-side cancelled the `trio`-side,\n'
+                f'c(>\n'
+                f'  |_{aio_task}\n\n'
+
+                f'{trio_err!r}\n'
             ) from aio_err
 
+        # maybe the chan-closure is due to something else?
         else:
             raise
 
-    finally:
+    except BaseException as _trio_err:
+        # await tractor.pause(shield=True)
+        trio_err = _trio_err
+        log.exception(
+            '`trio`-side task errored?'
+        )
+
+        entered: bool = await _debug._maybe_enter_pm(
+            trio_err,
+            api_frame=inspect.currentframe(),
+        )
         if (
-            # NOTE: always cancel the ``asyncio`` task if we've made it
-            # this far and it's not done.
-            not task.done() and aio_err
+            not entered
+            and
+            not is_multi_cancelled(trio_err)
+        ):
+            log.exception('actor crashed\n')
+
+        aio_taskc = AsyncioCancelled(
+            f'`trio`-side task errored!\n'
+            f'{trio_err}'
+        ) #from trio_err
+
+        try:
+            aio_task.set_exception(aio_taskc)
+        except (
+            asyncio.InvalidStateError,
+            RuntimeError,
+            # ^XXX, uhh bc apparently we can't use `.set_exception()`
+            # any more XD .. ??
+        ):
+            wait_on_aio_task = False
+
+        # import pdbp; pdbp.set_trace()
+        # raise aio_taskc from trio_err
+
+    finally:
+        # record wtv `trio`-side error transpired
+        chan._trio_err = trio_err
+        ya_trio_exited: bool = chan._trio_exited
+
+        # NOTE! by default always cancel the `asyncio` task if
+        # we've made it this far and it's not done.
+        # TODO, how to detect if there's an out-of-band error that
+        # caused the exit?
+        if (
+            cancel_aio_task_on_trio_exit
+            and
+            not aio_task.done()
+            and
+            aio_err
 
             # or the trio side has exited it's surrounding cancel scope
             # indicating the lifetime of the ``asyncio``-side task
             # should also be terminated.
-            or chan._trio_exited
-        ):
-            log.runtime(
-                f'Cancelling `asyncio`-task: {task.get_name()}'
+            or (
+                ya_trio_exited
+                and
+                not chan._trio_err   # XXX CRITICAL, `asyncio.Task.cancel()` is cucked man..
             )
-            # assert not aio_err, 'WTF how did asyncio do this?!'
-            task.cancel()
+        ):
+            report: str = (
+                'trio-side exited silently!'
+            )
+            assert not aio_err, 'WTF how did asyncio do this?!'
 
-        # Required to sync with the far end ``asyncio``-task to ensure
+            # if the `trio.Task` already exited the `open_channel_from()`
+            # block we ensure the asyncio-side gets signalled via an
+            # explicit exception and its `Queue` is shutdown.
+            if ya_trio_exited:
+                chan._to_aio.shutdown()
+
+                # 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'
+                            f'>>\n'
+                            f'|_{aio_task!r}\n'
+                        )
+                    )
+                else:
+                    aio_taskc_warn: str = (
+                        f'\n'
+                        f'MANUALLY Cancelling `asyncio`-task: {aio_task.get_name()}!\n\n'
+                        f'**THIS CAN SILENTLY SUPPRESS ERRORS FYI\n\n'
+                    )
+                    report += aio_taskc_warn
+                    # TODO XXX, figure out the case where calling this makes the
+                    # `test_infected_asyncio.py::test_trio_closes_early_and_channel_exits`
+                    # hang and then don't call it in that case!
+                    #
+                    aio_task.cancel(msg=aio_taskc_warn)
+
+            log.warning(report)
+
+        # Required to sync with the far end `asyncio`-task to ensure
         # any error is captured (via monkeypatching the
-        # ``channel._aio_err``) before calling ``maybe_raise_aio_err()``
+        # `channel._aio_err`) before calling ``maybe_raise_aio_err()``
         # below!
+        #
+        # XXX NOTE XXX the `task.set_exception(aio_taskc)` call above
+        # MUST NOT EXCEPT or this WILL HANG!!
+        #
+        # so if you get a hang maybe step through and figure out why
+        # it erroed out up there!
+        #
         if wait_on_aio_task:
+            # await chan.wait_aio_complete()
             await chan._aio_task_complete.wait()
+            log.info(
+                'asyncio-task is done and unblocked trio-side!\n'
+            )
+
+        # TODO?
+        # -[ ] make this a channel method, OR
+        # -[ ] just put back inline below?
+        #
+        def maybe_raise_aio_side_err(
+            trio_err: Exception,
+        ) -> None:
+            '''
+            Raise any `trio`-side-caused cancellation or legit task
+            error normally propagated from the caller of either,
+              - `open_channel_from()`
+              - `run_task()`
+
+            '''
+            aio_err: BaseException|None = chan._aio_err
+
+            # Check if the asyncio-side is the cause of the trio-side
+            # error.
+            if (
+                aio_err is not None
+                and
+                type(aio_err) is not AsyncioCancelled
+
+                # not isinstance(aio_err, CancelledError)
+                # type(aio_err) is not CancelledError
+            ):
+                # always raise from any captured asyncio error
+                if trio_err:
+                    raise trio_err from aio_err
+
+                raise aio_err
+
+            if trio_err:
+                raise trio_err
 
         # NOTE: if any ``asyncio`` error was caught, raise it here inline
         # here in the ``trio`` task
-        maybe_raise_aio_err()
+        # if trio_err:
+        maybe_raise_aio_side_err(
+            trio_err=trio_err
+        )
 
 
 async def run_task(
@@ -495,8 +815,8 @@ async def run_task(
 
 ) -> Any:
     '''
-    Run an `asyncio` async function or generator in a task, return
-    or stream the result back to `trio`.
+    Run an `asyncio`-compat async function or generator in a task,
+    return or stream the result back to `trio`.
 
     '''
     # simple async func
@@ -536,6 +856,7 @@ async def open_channel_from(
         provide_channels=True,
         **kwargs,
     )
+    # TODO, tuple form here?
     async with chan._from_aio:
         async with translate_aio_errors(
             chan,
@@ -544,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()
@@ -684,18 +1016,21 @@ def run_as_asyncio_guest(
     # Uh, oh.
     #
     # :o
-
-    # It looks like your event loop has caught a case of the ``trio``s.
-
-    # :()
-
-    # Don't worry, we've heard you'll barely notice. You might
-    # hallucinate a few more propagating errors and feel like your
-    # digestion has slowed but if anything get's too bad your parents
-    # will know about it.
-
+    #
+    # looks like your stdlib event loop has caught a case of "the trios" !
+    #
+    # :O
+    #
+    # Don't worry, we've heard you'll barely notice.
+    #
     # :)
-
+    #
+    # You might hallucinate a few more propagating errors and feel
+    # like your digestion has slowed, but if anything get's too bad
+    # your parents will know about it.
+    #
+    # B)
+    #
     async def aio_main(trio_main):
         '''
         Main `asyncio.Task` which calls
@@ -712,24 +1047,29 @@ def run_as_asyncio_guest(
             '-> built a `trio`-done future\n'
         )
 
-        # TODO: shoudn't this be done in the guest-run trio task?
-        # if debug_mode():
-        #     # XXX make it obvi we know this isn't supported yet!
-        #     log.error(
-        #         'Attempting to enter unsupported `greenback` init '
-        #         'from `asyncio` task..'
-        #     )
+        # TODO: is this evern run or needed?
+        # -[ ] pretty sure it never gets run for root-infected-aio
+        #     since this main task is always the parent of any
+        #     eventual `open_root_actor()` call?
+        if debug_mode():
+            log.error(
+                'Attempting to enter non-required `greenback` init '
+                'from `asyncio` task ???'
+            )
+            # XXX make it obvi we know this isn't supported yet!
+            assert 0
             # await _debug.maybe_init_greenback(
             #     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'
             )
 
+            # import pdbp; pdbp.set_trace()
             if isinstance(main_outcome, Error):
                 error: BaseException = main_outcome.error
 
@@ -750,7 +1090,7 @@ def run_as_asyncio_guest(
                 trio_done_fute.set_result(main_outcome)
 
             log.info(
-                f'`trio` guest-run finished with outcome\n'
+                f'`trio` guest-run finished with,\n'
                 f')>\n'
                 f'|_{trio_done_fute}\n'
             )
@@ -767,6 +1107,9 @@ def run_as_asyncio_guest(
             'Infecting `asyncio`-process with a `trio` guest-run!\n'
         )
 
+        # TODO, somehow bootstrap this!
+        _runtime_vars['_is_infected_aio'] = True
+
         trio.lowlevel.start_guest_run(
             trio_main,
             run_sync_soon_threadsafe=loop.call_soon_threadsafe,
@@ -775,6 +1118,18 @@ def run_as_asyncio_guest(
         fute_err: BaseException|None = None
         try:
             out: Outcome = await asyncio.shield(trio_done_fute)
+            # ^TODO still don't really understand why the `.shield()`
+            # is required ... ??
+            # https://docs.python.org/3/library/asyncio-task.html#asyncio.shield
+            # ^ seems as though in combo with the try/except here
+            # we're BOLDLY INGORING cancel of the trio fute?
+            #
+            # I guess it makes sense bc we don't want `asyncio` to
+            # cancel trio just because they can't handle SIGINT
+            # sanely? XD .. kk
+
+            # XXX, sin-shield causes guest-run abandons on SIGINT..
+            # out: Outcome = await trio_done_fute
 
             # NOTE will raise (via `Error.unwrap()`) from any
             # exception packed into the guest-run's `main_outcome`.
@@ -797,27 +1152,32 @@ def run_as_asyncio_guest(
             fute_err = _fute_err
             err_message: str = (
                 'main `asyncio` task '
+                'was cancelled!\n'
             )
-            if isinstance(fute_err, asyncio.CancelledError):
-                err_message += 'was cancelled!\n'
-            else:
-                err_message += f'errored with {out.error!r}\n'
 
+            # TODO, handle possible edge cases with
+            # `open_root_actor()` closing before this is run!
+            #
             actor: tractor.Actor = tractor.current_actor()
+
             log.exception(
                 err_message
                 +
                 'Cancelling `trio`-side `tractor`-runtime..\n'
-                f'c)>\n'
+                f'c(>\n'
                 f'  |_{actor}.cancel_soon()\n'
             )
 
-            # XXX WARNING XXX the next LOCs are super important, since
-            # without them, we can get guest-run abandonment cases
-            # where `asyncio` will not schedule or wait on the `trio`
-            # guest-run task before final shutdown! This is
-            # particularly true if the `trio` side has tasks doing
-            # shielded work when a SIGINT condition occurs.
+            # XXX WARNING XXX the next LOCs are super important!
+            #
+            # SINCE without them, we can get guest-run ABANDONMENT
+            # cases where `asyncio` will not schedule or wait on the
+            # guest-run `trio.Task` nor invoke its registered
+            # `trio_done_callback()` before final shutdown!
+            #
+            # This is particularly true if the `trio` side has tasks
+            # in shielded sections when an OC-cancel (SIGINT)
+            # condition occurs!
             #
             # We now have the
             # `test_infected_asyncio.test_sigint_closes_lifetime_stack()`
@@ -881,7 +1241,12 @@ def run_as_asyncio_guest(
 
             try:
                 return trio_done_fute.result()
-            except asyncio.exceptions.InvalidStateError as state_err:
+            except (
+                asyncio.InvalidStateError,
+                # asyncio.CancelledError,
+                # ^^XXX `.shield()` call above prevents this??
+
+            )as state_err:
 
                 # XXX be super dupere noisy about abandonment issues!
                 aio_task: asyncio.Task = asyncio.current_task()
@@ -905,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