12 changed files with 311 additions and 1266 deletions
--- a/tests/conftest.py
+++ b/tests/conftest.py
@ -150,18 +150,6 @@ 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)
--- a/tests/devx/test_pause_from_non_trio.py
+++ b/tests/devx/test_pause_from_non_trio.py
@ -218,9 +218,10 @@ def expect_any_of(
        )
    return expected_patts
    # yield child
-def test_sync_pause_from_aio_task(
+def test_pause_from_asyncio_task(
    spawn,
    ctlc: bool
    # ^TODO, fix for `asyncio`!!
@ -326,25 +327,3 @@ def test_sync_pause_from_aio_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
--- a/tests/test_infected_asyncio.py
+++ b/tests/test_infected_asyncio.py
@ -5,7 +5,6 @@ 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
@ -109,9 +108,7 @@ 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
@ -183,8 +180,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 tn:
+            async with tractor.open_nursery() as n:
-                await tn.run_in_actor(
+                await n.run_in_actor(
                    asyncio_actor,
                    target='aio_sleep_forever',
                    expect_err='trio.Cancelled',
@ -204,12 +201,9 @@ 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(
+            trio.open_nursery() as n,
-                    # TODO, for new `trio` / py3.13
+
                    # strict_exception_groups=False,
                ) as tn,
            tractor.to_asyncio.open_channel_from(
                sleep_and_err,
            ) as (first, chan),
@ -218,20 +212,12 @@ async def trio_ctx(
            assert first == 'start'
            # spawn another asyncio task for the cuck of it.
-                tn.start_soon(
+            n.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',
@ -249,6 +235,7 @@ 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(
@ -320,9 +307,7 @@ async def aio_cancel():
    await aio_sleep_forever()
-def test_aio_cancelled_from_aio_causes_trio_cancelled(
+def test_aio_cancelled_from_aio_causes_trio_cancelled(reg_addr):
    reg_addr: tuple,
 ):
    '''
    When the `asyncio.Task` cancels itself the `trio` side cshould
    also cancel and teardown and relay the cancellation cross-process
@ -419,7 +404,6 @@ 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
@ -438,15 +422,10 @@ 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
@ -456,19 +435,11 @@ 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 (
@ -489,8 +460,7 @@ 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(
@ -530,37 +500,19 @@ def test_trio_error_cancels_intertask_chan(reg_addr):
    excinfo.value.boxed_type is Exception
-def test_trio_closes_early_and_channel_exits(
+def test_trio_closes_early_and_channel_exits(reg_addr):
    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():
-        with trio.fail_after(2):
+        async with tractor.open_nursery() as n:
            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
-                print('waiting on final infected subactor result..')
+            await portal.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(
+    trio.run(main)
        main,
        # strict_exception_groups=False,
    )
 def test_aio_errors_and_channel_propagates_and_closes(reg_addr):
@ -584,6 +536,11 @@ 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,
@ -605,19 +562,15 @@ async def 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'
        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)
@ -696,6 +649,7 @@ def test_echoserver_detailed_mechanics(
        trio.run(main)
@tractor.context
 async def manage_file(
    ctx: tractor.Context,
--- a/tests/test_root_infect_asyncio.py
+++ b/tests/test_root_infect_asyncio.py
@ -1,244 +0,0 @@
 '''
 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]
--- a/tests/test_trioisms.py
+++ b/tests/test_trioisms.py
@ -3,10 +3,6 @@ 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
@ -84,115 +80,3 @@ 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
--- a/tractor/_exceptions.py
+++ b/tractor/_exceptions.py
@ -1146,51 +1146,19 @@ def unpack_error(
 def is_multi_cancelled(
-    exc: BaseException|BaseExceptionGroup,
+    exc: BaseException|BaseExceptionGroup
-
+) -> bool:
    ignore_nested: set[BaseException] = set(),
 ) -> bool|BaseExceptionGroup:
    '''
-    Predicate to determine if an `BaseExceptionGroup` only contains
+    Predicate to determine if a possible ``BaseExceptionGroup`` contains
-    some (maybe nested) set of sub-grouped exceptions (like only
+    only ``trio.Cancelled`` sub-exceptions (and is likely the result of
-    `trio.Cancelled`s which get swallowed silently by default) and is
+    cancelling a collection of subtasks.
    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):
-        matched_exc: BaseExceptionGroup|None = exc.subgroup(
+        return exc.subgroup(
-            tuple(ignore_nested),
+            lambda exc: isinstance(exc, trio.Cancelled)
        ) is not None
            # 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
--- a/tractor/_root.py
+++ b/tractor/_root.py
@ -95,13 +95,6 @@ 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,
@ -341,10 +334,6 @@ 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:
@ -386,7 +375,6 @@ 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
@ -395,15 +383,11 @@ 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(
+                    not is_multi_cancelled(err)
                        err,
                    )
                ):
                    logger.exception('Root actor crashed\n')
--- a/tractor/devx/_debug.py
+++ b/tractor/devx/_debug.py
@ -75,7 +75,6 @@ from tractor import _state
 from tractor._exceptions import (
    InternalError,
    NoRuntime,
    is_multi_cancelled,
 )
 from tractor._state import (
    current_actor,
@ -317,7 +316,6 @@ 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():
@ -445,10 +443,7 @@ class Lock:
                        f'|_{repl_task}\n'
                    )
            if message:
            log.devx(message)
            else:
                import pdbp; pdbp.set_trace()
        return we_released
@ -1748,7 +1743,7 @@ async def _pause(
    ] = trio.TASK_STATUS_IGNORED,
    **debug_func_kwargs,
-) -> tuple[Task, PdbREPL]|None:
+) -> tuple[PdbREPL, Task]|None:
    '''
    Inner impl for `pause()` to avoid the `trio.CancelScope.__exit__()`
    stack frame when not shielded (since apparently i can't figure out
@ -1934,7 +1929,7 @@ async def _pause(
                )
                with trio.CancelScope(shield=shield):
                    await trio.lowlevel.checkpoint()
-                return (repl, task)
+                return repl, task
            # elif repl_task:
            #     log.warning(
@ -2535,17 +2530,26 @@ 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_infected_aio: bool = actor.is_infected_aio()
+        is_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
@ -2561,14 +2565,20 @@ def pause_from_sync(
        if (
            not is_trio_thread
            and
-            not asyncio_task
+            not is_aio  # see below for this usage
        ):
            # 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..
-            repl_owner: Thread = thread
+            thread: threading.Thread = threading.current_thread()
            repl_owner = 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'
@ -2627,32 +2637,13 @@ 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_infected_aio  # as in, the special actor-runtime mode
+            is_aio
            # ^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(
@ -2669,53 +2660,11 @@ def pause_from_sync(
                    **_pause_kwargs
                )
            )
-                repl_owner = asyncio_task
+
-                bg_task, _ = greenback.await_(fute)
+            # TODO: for async version -> `.pause_from_aio()`?
                # TODO: ASYNC version -> `.pause_from_aio()`?
            # bg_task, _ = await fute
-
+            bg_task, _ = greenback.await_(fute)
-            # handle the case where an `asyncio` task has been
+            bg_task: asyncio.Task = asyncio.current_task()
            # 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
@ -2966,14 +2915,8 @@ 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()
@ -2990,8 +2933,7 @@ 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
+        and not is_multi_cancelled(err)
        debug_filter(err)
    ):
        api_frame: FrameType = api_frame or inspect.currentframe()
        tb: TracebackType = tb or sys.exc_info()[2]
@ -3172,7 +3114,7 @@ async def maybe_wait_for_debugger(
@cm
 def open_crash_handler(
    catch: set[BaseException] = {
-        # Exception,
+        Exception,
        BaseException,
    },
    ignore: set[BaseException] = {
@ -3193,30 +3135,14 @@ 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 boxed_maybe_exc
+        yield
    except tuple(catch) as err:
-        boxed_maybe_exc.value = err
+        if type(err) not in ignore:
-        if (
+
            type(err) not in ignore
            and
            not is_multi_cancelled(
                err,
                ignore_nested=ignore
            )
        ):
            try:
            # use our re-impl-ed version
            try:
                _post_mortem(
                    repl=mk_pdb(),
                    tb=sys.exc_info()[2],
@ -3224,13 +3150,13 @@ def open_crash_handler(
                )
            except bdb.BdbQuit:
                __tracebackhide__: bool = False
-                raise err
+                raise
            # XXX NOTE, `pdbp`'s version seems to lose the up-stack
            # tb-info?
            # pdbp.xpm()
-        raise err
+        raise
@cm
--- a/tractor/devx/pformat.py
+++ b/tractor/devx/pformat.py
@ -92,7 +92,7 @@ def pformat_boxed_tb(
        f' ------ {boxer_header} ------\n'
        f'{tb_body}'
        f' ------ {boxer_header}- ------\n'
-        f'_|'
+        f'_|\n'
    )
    tb_box_indent: str = (
        tb_box_indent
--- a/tractor/log.py
+++ b/tractor/log.py
@ -258,28 +258,20 @@ 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 = logger or logging.getLogger(_root_name)
+    log = rlog = logging.getLogger(_root_name)
    if (
        name
-        and
+        and name != _proj_name
        name != _proj_name
    ):
        # NOTE: for handling for modules that use ``get_logger(__name__)``
@ -314,10 +306,7 @@ 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(
+    logger = StackLevelAdapter(log, ActorContextInfo())
        log,
        ActorContextInfo(),
    )
    # additional levels
    for name, val in CUSTOM_LEVELS.items():
@ -331,24 +320,14 @@ 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.
    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.
    Yeah yeah, i know we can use ``DictConfig``. You do it.
    '''
-    log = get_logger(
+    log = get_logger(**kwargs)  # our root logger
-        logger=logger,
+    logger = log.logger
        **kwargs
    )  # set a root logger
    logger: Logger = log.logger
    if not level:
        return log
@ -367,13 +346,9 @@ def get_console_log(
            None,
        )
    ):
        fmt = LOG_FORMAT
        # if logger:
        #     fmt = None
        handler = StreamHandler()
        formatter = colorlog.ColoredFormatter(
-            fmt=fmt,
+            LOG_FORMAT,
            datefmt=DATE_FORMAT,
            log_colors=STD_PALETTE,
            secondary_log_colors=BOLD_PALETTE,
@ -390,7 +365,7 @@ def get_loglevel() -> str:
 # global module logger for tractor itself
-log: StackLevelAdapter = get_logger('tractor')
+log = get_logger('tractor')
 def at_least_level(
--- a/tractor/to_asyncio.py
+++ b/tractor/to_asyncio.py
@ -33,23 +33,12 @@ from typing import (
 )
 import tractor
-from tractor._exceptions import (
+from tractor._exceptions import AsyncioCancelled
    is_multi_cancelled,
 )
 from tractor._state import (
    debug_mode,
    _runtime_vars,
 )
 from tractor.devx import _debug
-from tractor.log import (
+from tractor.log import get_logger
    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,
@ -60,7 +49,7 @@ from outcome import (
    Outcome,
 )
-log: StackLevelAdapter = get_logger(__name__)
+log = get_logger(__name__)
 __all__ = [
@ -70,12 +59,7 @@ __all__ = [
@dataclass
-class LinkedTaskChannel(
+class LinkedTaskChannel(trio.abc.Channel):
    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``
@ -85,15 +69,12 @@ class LinkedTaskChannel(
    _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_cs: trio.CancelScope
    _aio_task_complete: trio.Event
    _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
@ -101,41 +82,7 @@ class LinkedTaskChannel(
    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,
@ -143,7 +90,21 @@ class LinkedTaskChannel(
            # being procesed.
            # wait_on_aio_task=False,
        ):
-                raise err
+
            # 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()
    async def send(self, item: Any) -> None:
        '''
@ -154,17 +115,6 @@ class LinkedTaskChannel(
        '''
        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
@ -204,6 +154,7 @@ class LinkedTaskChannel(
 def _run_asyncio_task(
    func: Callable,
    *,
    qsize: int = 1,
@ -213,9 +164,8 @@ def _run_asyncio_task(
 ) -> LinkedTaskChannel:
    '''
-    Run an `asyncio`-compat async function or generator in a task,
+    Run an ``asyncio`` async function or generator in a task, return
-    return or stream the result back to the caller
+    or stream the result back to the caller `trio.lowleve.Task`.
    `trio.lowleve.Task`.
    '''
    __tracebackhide__: bool = hide_tb
@ -248,30 +198,28 @@ def _run_asyncio_task(
    coro = func(**kwargs)
-    trio_task: trio.Task = trio.lowlevel.current_task()
+    cancel_scope = trio.CancelScope()
    trio_cs = trio.CancelScope()
    aio_task_complete = trio.Event()
    aio_err: BaseException|None = None
    chan = LinkedTaskChannel(
-        _to_aio=aio_q,  # asyncio.Queue
+        aio_q,  # asyncio.Queue
-        _from_aio=from_aio,  # recv chan
+        from_aio,  # recv chan
-        _to_trio=to_trio,  # send chan
+        to_trio,  # send chan
-        _trio_cs=trio_cs,
+
-        _trio_task=trio_task,
+        cancel_scope,
-        _aio_task_complete=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
@ -294,10 +242,8 @@ def _run_asyncio_task(
        else:
            if (
-                result != orig
+                result != orig and
-                and
+                aio_err is None and
                aio_err is None
                and
                # in the `open_channel_from()` case we don't
                # relay through the "return value".
@ -306,9 +252,6 @@ 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:
@ -317,19 +260,11 @@ def _run_asyncio_task(
                to_trio.close()
            aio_task_complete.set()
-            # await asyncio.sleep(0.1)
+            log.runtime(f'`asyncio` task: {task.get_name()} is complete')
            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(
+        raise TypeError(f"No support for invoking {coro}")
            f'Pass the async-fn NOT a coroutine\n'
            f'{coro!r}'
        )
    task: asyncio.Task = asyncio.create_task(
        wait_on_coro_final_result(
@ -353,55 +288,28 @@ 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(
+    def cancel_trio(task: asyncio.Task) -> None:
        task: asyncio.Task,
    ) -> None:
        '''
-        Cancel the parent `trio` task on any error raised by the
+        Cancel the calling `trio` task on error.
        `asyncio` side.
        '''
        nonlocal chan
-        relayed_aio_err: BaseException|None = chan._aio_err
+        aio_err: BaseException|None = chan._aio_err
-        aio_err: BaseException|None = None
+        task_err: BaseException|None = None
        # only to avoid `asyncio` complaining about uncaptured
        # task exceptions
        try:
            res: Any = task.result()
-            log.info(
+        except BaseException as terr:
-                '`trio` received final result from {task}\n'
+            task_err: BaseException = terr
                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 = (
-                '`trio`-side reports that the `asyncio`-side '
+                'Infected `asyncio` task {etype_str}\n'
                '{etype_str}\n'
                # ^NOTE filled in below
            )
-            if isinstance(_aio_err, CancelledError):
+            if isinstance(terr, CancelledError):
                msg += (
                    f'c)>\n'
                    f' |_{task}\n'
@ -418,143 +326,75 @@ def _run_asyncio_task(
                    msg.format(etype_str='errored')
                )
-        trio_err: BaseException|None = chan._trio_err
+            assert type(terr) is type(aio_err), (
-        if (
+                '`asyncio` task error mismatch?!?'
-            relayed_aio_err
+            )
-            or
+
-            trio_err
+        if aio_err is not None:
-        ):
+            # XXX: uhh is this true?
-            # import pdbp; pdbp.set_trace()
+            # 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`
+            # of error relay (at least in the ``asyncio.CancelledError``
-            # case) since we have no way to directly trigger a `trio`
+            # 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 in log-msg?
+            # from each side more pedantically?
            # elif (
            #     type(aio_err) is CancelledError
            #     and  # trio was the cause?
-            #     trio_cs.cancel_called
+            #     cancel_scope.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
+            # XXX: if not already, alway cancel the scope
-            # error in case the trio task is blocking on
+            # on a task error in case the trio task is blocking on
            # a checkpoint.
            cancel_scope.cancel()
            if (
-                not trio_cs.cancelled_caught
+                task_err
-                or
+                and
-                not trio_cs.cancel_called
+                aio_err is not task_err
            ):
-                # import pdbp; pdbp.set_trace()
+                raise aio_err from task_err
                trio_cs.cancel()
            # 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 relayed_aio_err:
                    raise aio_err from relayed_aio_err
            # raise any `asyncio` side error.
            raise aio_err
        log.info(
            '`trio` received final result from {task}\n'
            f'|_{res}\n'
        )
        # TODO: do we need this?
        # if task_err:
        #     cancel_scope.cancel()
        #     raise task_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]:
    '''
-    An error handling to cross-loop propagation context around
+    Error handling context around ``asyncio`` task spawns which
    `asyncio.Task` spawns via one of this module's APIs:
    - `open_channel_from()`
    - `run_task()`
    appropriately translates errors and cancels into ``trio`` land.
    '''
@ -562,248 +402,88 @@ async def translate_aio_errors(
    aio_err: BaseException|None = None
-    aio_task: asyncio.Task = chan._aio_task
+    # TODO: make thisi a channel method?
-    assert aio_task
+    def maybe_raise_aio_err(
-    trio_err: BaseException|None = None
+        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
    try:
-        yield  # back to one of the cross-loop apis
+        yield
    except trio.Cancelled as taskc:
        trio_err = taskc
        # should NEVER be the case that `trio` is cancel-handling
        # BEFORE the other side's task-ref was set!?
        assert chan._aio_task
        # 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 (
        trio.Cancelled,
    ):
        # relay cancel through to called ``asyncio`` task
        assert chan._aio_task
        chan._aio_task.cancel(
            msg=f'the `trio` caller task was cancelled: {trio_task.name}'
        )
        raise
    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 (
-            # aio-side is cancelled?
+            task.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'asyncio`-side cancelled the `trio`-side,\n'
+                f'Task cancelled\n'
-                f'c(>\n'
+                f'|_{task}\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
    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 (
            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
+            # NOTE: always cancel the ``asyncio`` task if we've made it
-            and
+            # this far and it's not done.
-            not aio_task.done()
+            not task.done() and aio_err
            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 (
+            or chan._trio_exited
                ya_trio_exited
                and
                not chan._trio_err   # XXX CRITICAL, `asyncio.Task.cancel()` is cucked man..
            )
        ):
-            report: str = (
+            log.runtime(
-                'trio-side exited silently!'
+                f'Cancelling `asyncio`-task: {task.get_name()}'
            )
-            assert not aio_err, 'WTF how did asyncio do this?!'
+            # assert not aio_err, 'WTF how did asyncio do this?!'
            task.cancel()
-            # if the `trio.Task` already exited the `open_channel_from()`
+        # Required to sync with the far end ``asyncio``-task to ensure
            # 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
-        # if trio_err:
+        maybe_raise_aio_err()
        maybe_raise_aio_side_err(
            trio_err=trio_err
        )
 async def run_task(
@ -815,8 +495,8 @@ async def run_task(
 ) -> Any:
    '''
-    Run an `asyncio`-compat async function or generator in a task,
+    Run an `asyncio` async function or generator in a task, return
-    return or stream the result back to `trio`.
+    or stream the result back to `trio`.
    '''
    # simple async func
@ -856,7 +536,6 @@ async def open_channel_from(
        provide_channels=True,
        **kwargs,
    )
    # TODO, tuple form here?
    async with chan._from_aio:
        async with translate_aio_errors(
            chan,
@ -865,22 +544,11 @@ async def open_channel_from(
            # sync to a "started()"-like first delivered value from the
            # ``asyncio`` task.
            try:
-                with (cs := chan._trio_cs):
+                with 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()
@ -1016,21 +684,18 @@ def run_as_asyncio_guest(
    # Uh, oh.
    #
    # :o
-    #
+
-    # looks like your stdlib event loop has caught a case of "the trios" !
+    # It looks like your event loop has caught a case of the ``trio``s.
-    #
+
-    # :O
+    # :()
-    #
+
-    # Don't worry, we've heard you'll barely notice.
+    # 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.
    # :)
-    #
+
    # 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
@ -1047,29 +712,24 @@ def run_as_asyncio_guest(
            '-> built a `trio`-done future\n'
        )
-        # TODO: is this evern run or needed?
+        # TODO: shoudn't this be done in the guest-run trio task?
-        # -[ ] pretty sure it never gets run for root-infected-aio
+        # if debug_mode():
-        #     since this main task is always the parent of any
+        #     # XXX make it obvi we know this isn't supported yet!
-        #     eventual `open_root_actor()` call?
+        #     log.error(
-        if debug_mode():
+        #         'Attempting to enter unsupported `greenback` init '
-            log.error(
+        #         'from `asyncio` task..'
-                '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: Outcome):
+        def trio_done_callback(main_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
@ -1090,7 +750,7 @@ def run_as_asyncio_guest(
                trio_done_fute.set_result(main_outcome)
            log.info(
-                f'`trio` guest-run finished with,\n'
+                f'`trio` guest-run finished with outcome\n'
                f')>\n'
                f'|_{trio_done_fute}\n'
            )
@ -1107,9 +767,6 @@ 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,
@ -1118,18 +775,6 @@ 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`.
@ -1152,32 +797,27 @@ 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!
+            # XXX WARNING XXX the next LOCs are super important, since
-            #
+            # without them, we can get guest-run abandonment cases
-            # SINCE without them, we can get guest-run ABANDONMENT
+            # where `asyncio` will not schedule or wait on the `trio`
-            # cases where `asyncio` will not schedule or wait on the
+            # guest-run task before final shutdown! This is
-            # guest-run `trio.Task` nor invoke its registered
+            # particularly true if the `trio` side has tasks doing
-            # `trio_done_callback()` before final shutdown!
+            # shielded work when a SIGINT condition occurs.
            #
            # 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()`
@ -1241,12 +881,7 @@ def run_as_asyncio_guest(
            try:
                return trio_done_fute.result()
-            except (
+            except asyncio.exceptions.InvalidStateError as state_err:
                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()
@ -1270,10 +905,6 @@ 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