Add a `tests/test_root_infect_asyncio`

Might as well break apart the specific test set since there are some
(minor) subtleties and the orig test mod is already getting pretty big
XD

Includes both the new "independent"-event-loops test as well as the std
usage base case suite.

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,22 +204,33 @@ async def trio_ctx(
     # this will block until the ``asyncio`` task sends a "first"
     # message.
     with trio.fail_after(2):
-        async with (
-            trio.open_nursery() as n,
+        try:
+            async with (
+                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),
+            ):
 
-            tractor.to_asyncio.open_channel_from(
-                sleep_and_err,
-            ) as (first, chan),
-        ):
+                assert first == 'start'
 
-            assert first == 'start'
+                # spawn another asyncio task for the cuck of it.
+                tn.start_soon(
+                    tractor.to_asyncio.run_task,
+                    aio_sleep_forever,
+                )
+                await trio.sleep_forever()
 
-            # spawn another asyncio task for the cuck of it.
-            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(
@@ -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:
-            portal = await n.run_in_actor(
-                stream_from_aio,
-                exit_early=True,
-                infect_asyncio=True,
-            )
-            # should raise RAE diectly
-            await portal.result()
+        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
+                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,41 +584,40 @@ def test_aio_errors_and_channel_propagates_and_closes(reg_addr):
     excinfo.value.boxed_type is Exception
 
 
+async def aio_echo_server(
+    to_trio: trio.MemorySendChannel,
+    from_trio: asyncio.Queue,
+) -> None:
+
+    to_trio.send_nowait('start')
+
+    while True:
+        msg = await from_trio.get()
+
+        # echo the msg back
+        to_trio.send_nowait(msg)
+
+        # if we get the terminate sentinel
+        # break the echo loop
+        if msg is None:
+            print('breaking aio echo loop')
+            break
+
+    print('exiting asyncio task')
+
+
 @tractor.context
 async def trio_to_aio_echo_server(
-    ctx: tractor.Context,
+    ctx: tractor.Context|None,
 ):
-
-    async def aio_echo_server(
-        to_trio: trio.MemorySendChannel,
-        from_trio: asyncio.Queue,
-    ) -> None:
-
-        to_trio.send_nowait('start')
-
-        while True:
-            msg = await from_trio.get()
-
-            # echo the msg back
-            to_trio.send_nowait(msg)
-
-            # if we get the terminate sentinel
-            # break the echo loop
-            if msg is None:
-                print('breaking aio echo loop')
-                break
-
-        print('exiting asyncio task')
-
     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/_context.py

@@ -47,6 +47,9 @@ from functools import partial
 import inspect
 from pprint import pformat
 import textwrap
+from types import (
+    UnionType,
+)
 from typing import (
     Any,
     AsyncGenerator,
@@ -2544,7 +2547,14 @@ def context(
     name: str
     param: Type
     for name, param in annots.items():
-        if param is Context:
+        if (
+            param is Context
+            or (
+                isinstance(param, UnionType)
+                and
+                Context in param.__args__
+            )
+        ):
             ctx_var_name: str = name
             break
     else:

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'
                     )
 
-            log.devx(message)
+            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,24 +2561,18 @@ 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'
-                    )
+            if is_root:
+                message += (
+                    f'-> called from a root-actor bg {thread}\n'
+                )
 
                 message += (
                     '-> scheduling `._pause_from_bg_root_thread()`..\n'
@@ -2637,34 +2627,95 @@ 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()
-            DebugStatus.shield_sigint()
-            fute: asyncio.Future = run_trio_task_in_future(
-                partial(
-                    _pause,
-                    debug_func=None,
-                    repl=repl,
-                    hide_tb=hide_tb,
 
-                    # XXX to prevent `._pause()` for setting
-                    # `DebugStatus.repl_task` to the gb task!
-                    called_from_sync=True,
-                    called_from_bg_thread=True,
+            if greenback.has_portal():
+                DebugStatus.shield_sigint()
+                fute: asyncio.Future = run_trio_task_in_future(
+                    partial(
+                        _pause,
+                        debug_func=None,
+                        repl=repl,
+                        hide_tb=hide_tb,
 
-                    **_pause_kwargs
+                        # XXX to prevent `._pause()` for setting
+                        # `DebugStatus.repl_task` to the gb task!
+                        called_from_sync=True,
+                        called_from_bg_thread=True,
+
+                        **_pause_kwargs
+                    )
                 )
-            )
+                repl_owner = asyncio_task
+                bg_task, _ = greenback.await_(fute)
+                # TODO: ASYNC version -> `.pause_from_aio()`?
+                # bg_task, _ = await fute
 
-            # TODO: for async version -> `.pause_from_aio()`?
-            # bg_task, _ = await fute
-            bg_task, _ = greenback.await_(fute)
-            bg_task: asyncio.Task = asyncio.current_task()
+            # 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/hilevel/__init__.py

@@ -0,0 +1,26 @@
+# tractor: structured concurrent "actors".
+# Copyright 2024-eternity Tyler Goodlet.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+High level design patterns, APIs and runtime extensions built on top
+of the `tractor` runtime core.
+
+'''
+from ._service import (
+    open_service_mngr as open_service_mngr,
+    get_service_mngr as get_service_mngr,
+    ServiceMngr as ServiceMngr,
+)

tractor/hilevel/_service.py

@@ -0,0 +1,592 @@
+# tractor: structured concurrent "actors".
+# Copyright 2024-eternity Tyler Goodlet.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+Daemon subactor as service(s) management and supervision primitives
+and API.
+
+'''
+from __future__ import annotations
+from contextlib import (
+    asynccontextmanager as acm,
+    # contextmanager as cm,
+)
+from collections import defaultdict
+from dataclasses import (
+    dataclass,
+    field,
+)
+import functools
+import inspect
+from typing import (
+    Callable,
+    Any,
+)
+
+import tractor
+import trio
+from trio import TaskStatus
+from tractor import (
+    log,
+    ActorNursery,
+    current_actor,
+    ContextCancelled,
+    Context,
+    Portal,
+)
+
+log = log.get_logger('tractor')
+
+
+# TODO: implement a `@singleton` deco-API for wrapping the below
+# factory's impl for general actor-singleton use?
+#
+# -[ ] go through the options peeps on SO did?
+#  * https://stackoverflow.com/questions/6760685/what-is-the-best-way-of-implementing-singleton-in-python
+#  * including @mikenerone's answer
+#   |_https://stackoverflow.com/questions/6760685/what-is-the-best-way-of-implementing-singleton-in-python/39186313#39186313
+#
+# -[ ] put it in `tractor.lowlevel._globals` ?
+#  * fits with our oustanding actor-local/global feat req?
+#   |_ https://github.com/goodboy/tractor/issues/55
+#  * how can it relate to the `Actor.lifetime_stack` that was
+#    silently patched in?
+#   |_ we could implicitly call both of these in the same
+#     spot in the runtime using the lifetime stack?
+#    - `open_singleton_cm().__exit__()`
+#    -`del_singleton()`
+#   |_ gives SC fixtue semantics to sync code oriented around
+#     sub-process lifetime?
+#  * what about with `trio.RunVar`?
+#   |_https://trio.readthedocs.io/en/stable/reference-lowlevel.html#trio.lowlevel.RunVar
+#    - which we'll need for no-GIL cpython (right?) presuming
+#      multiple `trio.run()` calls in process?
+#
+#
+# @singleton
+# async def open_service_mngr(
+#     **init_kwargs,
+# ) -> ServiceMngr:
+#     '''
+#     Note this function body is invoke IFF no existing singleton instance already
+#     exists in this proc's memory.
+
+#     '''
+#     # setup
+#     yield ServiceMngr(**init_kwargs)
+#     # teardown
+
+
+# a deletion API for explicit instance de-allocation?
+# @open_service_mngr.deleter
+# def del_service_mngr() -> None:
+#     mngr = open_service_mngr._singleton[0]
+#     open_service_mngr._singleton[0] = None
+#     del mngr
+
+
+
+# TODO: implement a singleton deco-API for wrapping the below
+# factory's impl for general actor-singleton use?
+#
+# @singleton
+# async def open_service_mngr(
+#     **init_kwargs,
+# ) -> ServiceMngr:
+#     '''
+#     Note this function body is invoke IFF no existing singleton instance already
+#     exists in this proc's memory.
+
+#     '''
+#     # setup
+#     yield ServiceMngr(**init_kwargs)
+#     # teardown
+
+
+
+# TODO: singleton factory API instead of a class API
+@acm
+async def open_service_mngr(
+    *,
+    debug_mode: bool = False,
+
+    # NOTE; since default values for keyword-args are effectively
+    # module-vars/globals as per the note from,
+    # https://docs.python.org/3/tutorial/controlflow.html#default-argument-values
+    #
+    # > "The default value is evaluated only once. This makes
+    #   a difference when the default is a mutable object such as
+    #   a list, dictionary, or instances of most classes"
+    #
+    _singleton: list[ServiceMngr|None] = [None],
+    **init_kwargs,
+
+) -> ServiceMngr:
+    '''
+    Open an actor-global "service-manager" for supervising a tree
+    of subactors and/or actor-global tasks.
+
+    The delivered `ServiceMngr` is singleton instance for each
+    actor-process, that is, allocated on first open and never
+    de-allocated unless explicitly deleted by al call to
+    `del_service_mngr()`.
+
+    '''
+    # TODO: factor this an allocation into
+    # a `._mngr.open_service_mngr()` and put in the
+    # once-n-only-once setup/`.__aenter__()` part!
+    # -[ ] how to make this only happen on the `mngr == None` case?
+    #  |_ use `.trionics.maybe_open_context()` (for generic
+    #     async-with-style-only-once of the factory impl, though
+    #     what do we do for the allocation case?
+    #    / `.maybe_open_nursery()` (since for this specific case
+    #    it's simpler?) to activate
+    async with (
+        tractor.open_nursery() as an,
+        trio.open_nursery() as tn,
+    ):
+        # impl specific obvi..
+        init_kwargs.update({
+            'an': an,
+            'tn': tn,
+        })
+
+        mngr: ServiceMngr|None
+        if (mngr := _singleton[0]) is None:
+
+            log.info('Allocating a new service mngr!')
+            mngr = _singleton[0] = ServiceMngr(**init_kwargs)
+
+            # TODO: put into `.__aenter__()` section of
+            # eventual `@singleton_acm` API wrapper.
+            #
+            # assign globally for future daemon/task creation
+            mngr.an = an
+            mngr.tn = tn
+
+        else:
+            assert (mngr.an and mngr.tn)
+            log.info(
+                'Using extant service mngr!\n\n'
+                f'{mngr!r}\n'  # it has a nice `.__repr__()` of services state
+            )
+
+        try:
+            # NOTE: this is a singleton factory impl specific detail
+            # which should be supported in the condensed
+            # `@singleton_acm` API?
+            mngr.debug_mode = debug_mode
+
+            yield mngr
+        finally:
+            # TODO: is this more clever/efficient?
+            # if 'samplerd' in mngr.service_ctxs:
+            #     await mngr.cancel_service('samplerd')
+            tn.cancel_scope.cancel()
+
+
+
+def get_service_mngr() -> ServiceMngr:
+    '''
+    Try to get the singleton service-mngr for this actor presuming it
+    has already been allocated using,
+
+    .. code:: python
+
+        async with open_<@singleton_acm(func)>() as mngr`
+            ... this block kept open ...
+
+    If not yet allocated raise a `ServiceError`.
+
+    '''
+    # https://stackoverflow.com/a/12627202
+    # https://docs.python.org/3/library/inspect.html#inspect.Signature
+    maybe_mngr: ServiceMngr|None = inspect.signature(
+        open_service_mngr
+    ).parameters['_singleton'].default[0]
+
+    if maybe_mngr is None:
+        raise RuntimeError(
+            'Someone must allocate a `ServiceMngr` using\n\n'
+            '`async with open_service_mngr()` beforehand!!\n'
+        )
+
+    return maybe_mngr
+
+
+async def _open_and_supervise_service_ctx(
+    serman: ServiceMngr,
+    name: str,
+    ctx_fn: Callable,  # TODO, type for `@tractor.context` requirement
+    portal: Portal,
+
+    allow_overruns: bool = False,
+    task_status: TaskStatus[
+        tuple[
+            trio.CancelScope,
+            Context,
+            trio.Event,
+            Any,
+        ]
+    ] = trio.TASK_STATUS_IGNORED,
+    **ctx_kwargs,
+
+) -> Any:
+    '''
+    Open a remote IPC-context defined by `ctx_fn` in the
+    (service) actor accessed via `portal` and supervise the
+    (local) parent task to termination at which point the remote
+    actor runtime is cancelled alongside it.
+
+    The main application is for allocating long-running
+    "sub-services" in a main daemon and explicitly controlling
+    their lifetimes from an actor-global singleton.
+
+    '''
+    # TODO: use the ctx._scope directly here instead?
+    # -[ ] actually what semantics do we expect for this
+    #   usage!?
+    with trio.CancelScope() as cs:
+        try:
+            async with portal.open_context(
+                ctx_fn,
+                allow_overruns=allow_overruns,
+                **ctx_kwargs,
+
+            ) as (ctx, started):
+
+                # unblock once the remote context has started
+                complete = trio.Event()
+                task_status.started((
+                    cs,
+                    ctx,
+                    complete,
+                    started,
+                ))
+                log.info(
+                    f'`pikerd` service {name} started with value {started}'
+                )
+                # wait on any context's return value
+                # and any final portal result from the
+                # sub-actor.
+                ctx_res: Any = await ctx.wait_for_result()
+
+                # NOTE: blocks indefinitely until cancelled
+                # either by error from the target context
+                # function or by being cancelled here by the
+                # surrounding cancel scope.
+                return (
+                    await portal.wait_for_result(),
+                    ctx_res,
+                )
+
+        except ContextCancelled as ctxe:
+            canceller: tuple[str, str] = ctxe.canceller
+            our_uid: tuple[str, str] = current_actor().uid
+            if (
+                canceller != portal.chan.uid
+                and
+                canceller != our_uid
+            ):
+                log.cancel(
+                    f'Actor-service `{name}` was remotely cancelled by a peer?\n'
+
+                    # TODO: this would be a good spot to use
+                    # a respawn feature Bo
+                    f'-> Keeping `pikerd` service manager alive despite this inter-peer cancel\n\n'
+
+                    f'cancellee: {portal.chan.uid}\n'
+                    f'canceller: {canceller}\n'
+                )
+            else:
+                raise
+
+        finally:
+            # NOTE: the ctx MUST be cancelled first if we
+            # don't want the above `ctx.wait_for_result()` to
+            # raise a self-ctxc. WHY, well since from the ctx's
+            # perspective the cancel request will have
+            # arrived out-out-of-band at the `Actor.cancel()`
+            # level, thus `Context.cancel_called == False`,
+            # meaning `ctx._is_self_cancelled() == False`.
+            # with trio.CancelScope(shield=True):
+            # await ctx.cancel()
+            await portal.cancel_actor()  # terminate (remote) sub-actor
+            complete.set()  # signal caller this task is done
+            serman.service_ctxs.pop(name)  # remove mngr entry
+
+
+# TODO: we need remote wrapping and a general soln:
+# - factor this into a ``tractor.highlevel`` extension # pack for the
+#   library.
+# - wrap a "remote api" wherein you can get a method proxy
+#   to the pikerd actor for starting services remotely!
+# - prolly rename this to ActorServicesNursery since it spawns
+#   new actors and supervises them to completion?
+@dataclass
+class ServiceMngr:
+    '''
+    A multi-subactor-as-service manager.
+
+    Spawn, supervise and monitor service/daemon subactors in a SC
+    process tree.
+
+    '''
+    an: ActorNursery
+    tn: trio.Nursery
+    debug_mode: bool = False # tractor sub-actor debug mode flag
+
+    service_tasks: dict[
+        str,
+        tuple[
+            trio.CancelScope,
+            trio.Event,
+        ]
+    ] = field(default_factory=dict)
+
+    service_ctxs: dict[
+        str,
+        tuple[
+            trio.CancelScope,
+            Context,
+            Portal,
+            trio.Event,
+        ]
+    ] = field(default_factory=dict)
+
+    # internal per-service task mutexs
+    _locks = defaultdict(trio.Lock)
+
+    # TODO, unify this interface with our `TaskManager` PR!
+    #
+    #
+    async def start_service_task(
+        self,
+        name: str,
+        # TODO: typevar for the return type of the target and then
+        # use it below for `ctx_res`?
+        fn: Callable,
+
+        allow_overruns: bool = False,
+        **ctx_kwargs,
+
+    ) -> tuple[
+        trio.CancelScope,
+        Any,
+        trio.Event,
+    ]:
+        async def _task_manager_start(
+            task_status: TaskStatus[
+                tuple[
+                    trio.CancelScope,
+                    trio.Event,
+                ]
+            ] = trio.TASK_STATUS_IGNORED,
+        ) -> Any:
+
+            task_cs = trio.CancelScope()
+            task_complete = trio.Event()
+
+            with task_cs as cs:
+                task_status.started((
+                    cs,
+                    task_complete,
+                ))
+                try:
+                    await fn()
+                except trio.Cancelled as taskc:
+                    log.cancel(
+                        f'Service task for `{name}` was cancelled!\n'
+                        # TODO: this would be a good spot to use
+                        # a respawn feature Bo
+                    )
+                    raise taskc
+                finally:
+                    task_complete.set()
+        (
+            cs,
+            complete,
+        ) = await self.tn.start(_task_manager_start)
+
+        # store the cancel scope and portal for later cancellation or
+        # retstart if needed.
+        self.service_tasks[name] = (
+            cs,
+            complete,
+        )
+        return (
+            cs,
+            complete,
+        )
+
+    async def cancel_service_task(
+        self,
+        name: str,
+
+    ) -> Any:
+        log.info(f'Cancelling `pikerd` service {name}')
+        cs, complete = self.service_tasks[name]
+
+        cs.cancel()
+        await complete.wait()
+        # TODO, if we use the `TaskMngr` from #346
+        # we can also get the return value from the task!
+
+        if name in self.service_tasks:
+            # TODO: custom err?
+            # raise ServiceError(
+            raise RuntimeError(
+                f'Service task {name!r} not terminated!?\n'
+            )
+
+    async def start_service_ctx(
+        self,
+        name: str,
+        portal: Portal,
+        # TODO: typevar for the return type of the target and then
+        # use it below for `ctx_res`?
+        ctx_fn: Callable,
+        **ctx_kwargs,
+
+    ) -> tuple[
+        trio.CancelScope,
+        Context,
+        Any,
+    ]:
+        '''
+        Start a remote IPC-context defined by `ctx_fn` in a background
+        task and immediately return supervision primitives to manage it:
+
+        - a `cs: CancelScope` for the newly allocated bg task
+        - the `ipc_ctx: Context` to manage the remotely scheduled
+          `trio.Task`.
+        - the `started: Any` value returned by the remote endpoint
+          task's `Context.started(<value>)` call.
+
+        The bg task supervises the ctx such that when it terminates the supporting
+        actor runtime is also cancelled, see `_open_and_supervise_service_ctx()`
+        for details.
+
+        '''
+        cs, ipc_ctx, complete, started = await self.tn.start(
+            functools.partial(
+                _open_and_supervise_service_ctx,
+                serman=self,
+                name=name,
+                ctx_fn=ctx_fn,
+                portal=portal,
+                **ctx_kwargs,
+            )
+        )
+
+        # store the cancel scope and portal for later cancellation or
+        # retstart if needed.
+        self.service_ctxs[name] = (cs, ipc_ctx, portal, complete)
+        return (
+            cs,
+            ipc_ctx,
+            started,
+        )
+
+    async def start_service(
+        self,
+        daemon_name: str,
+        ctx_ep: Callable,  # kwargs must `partial`-ed in!
+        # ^TODO, type for `@tractor.context` deco-ed funcs!
+
+        debug_mode: bool = False,
+        **start_actor_kwargs,
+
+    ) -> Context:
+        '''
+        Start new subactor and schedule a supervising "service task"
+        in it which explicitly defines the sub's lifetime.
+
+        "Service daemon subactors" are cancelled (and thus
+        terminated) using the paired `.cancel_service()`.
+
+        Effectively this API can be used to manage "service daemons"
+        spawned under a single parent actor with supervision
+        semantics equivalent to a one-cancels-one style actor-nursery
+        or "(subactor) task manager" where each subprocess's (and
+        thus its embedded actor runtime) lifetime is synced to that
+        of the remotely spawned task defined by `ctx_ep`.
+
+        The funcionality can be likened to a "daemonized" version of
+        `.hilevel.worker.run_in_actor()` but with supervision
+        controls offered by `tractor.Context` where the main/root
+        remotely scheduled `trio.Task` invoking `ctx_ep` determines
+        the underlying subactor's lifetime.
+
+        '''
+        entry: tuple|None = self.service_ctxs.get(daemon_name)
+        if entry:
+            (cs, sub_ctx, portal, complete) = entry
+            return sub_ctx
+
+        if daemon_name not in self.service_ctxs:
+            portal: Portal = await self.an.start_actor(
+                daemon_name,
+                debug_mode=(  # maybe set globally during allocate
+                    debug_mode
+                    or
+                    self.debug_mode
+                ),
+                **start_actor_kwargs,
+            )
+            ctx_kwargs: dict[str, Any] = {}
+            if isinstance(ctx_ep, functools.partial):
+                ctx_kwargs: dict[str, Any] = ctx_ep.keywords
+                ctx_ep: Callable = ctx_ep.func
+
+            (
+                cs,
+                sub_ctx,
+                started,
+            ) = await self.start_service_ctx(
+                name=daemon_name,
+                portal=portal,
+                ctx_fn=ctx_ep,
+                **ctx_kwargs,
+            )
+
+            return sub_ctx
+
+    async def cancel_service(
+        self,
+        name: str,
+
+    ) -> Any:
+        '''
+        Cancel the service task and actor for the given ``name``.
+
+        '''
+        log.info(f'Cancelling `pikerd` service {name}')
+        cs, sub_ctx, portal, complete = self.service_ctxs[name]
+
+        # cs.cancel()
+        await sub_ctx.cancel()
+        await complete.wait()
+
+        if name in self.service_ctxs:
+            # TODO: custom err?
+            # raise ServiceError(
+            raise RuntimeError(
+                f'Service actor for {name} not terminated and/or unknown?'
+            )
+
+        # assert name not in self.service_ctxs, \
+        #     f'Serice task for {name} not terminated?'

tractor/log.py

@@ -258,20 +258,28 @@ class ActorContextInfo(Mapping):
 
 
 def get_logger(
-
-    name: str | None = None,
+    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__)``
@@ -283,7 +291,7 @@ def get_logger(
         #   since in python the {filename} is always this same
         #   module-file.
 
-        sub_name: None | str = None
+        sub_name: None|str = None
         rname, _, sub_name = name.partition('.')
         pkgpath, _, modfilename = sub_name.rpartition('.')
 
@@ -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():
@@ -319,15 +330,25 @@ def get_logger(
 
 
 def get_console_log(
-    level: str | None = None,
+    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,19 @@ from typing import (
 )
 
 import tractor
-from tractor._exceptions import AsyncioCancelled
+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 get_logger
+from tractor.log import (
+    get_logger,
+    StackLevelAdapter,
+)
 from tractor.trionics._broadcast import (
     broadcast_receiver,
     BroadcastReceiver,
@@ -49,7 +56,7 @@ from outcome import (
     Outcome,
 )
 
-log = get_logger(__name__)
+log: StackLevelAdapter = get_logger(__name__)
 
 
 __all__ = [
@@ -69,9 +76,10 @@ class LinkedTaskChannel(trio.abc.Channel):
     _to_aio: asyncio.Queue
     _from_aio: trio.MemoryReceiveChannel
     _to_trio: trio.MemorySendChannel
-
     _trio_cs: trio.CancelScope
     _aio_task_complete: trio.Event
+
+    _trio_err: BaseException|None = None
     _trio_exited: bool = False
 
     # set after ``asyncio.create_task()``
@@ -83,28 +91,40 @@ class LinkedTaskChannel(trio.abc.Channel):
         await self._from_aio.aclose()
 
     async def receive(self) -> Any:
-        async with translate_aio_errors(
-            self,
-
-            # XXX: obviously this will deadlock if an on-going stream is
-            # being procesed.
-            # wait_on_aio_task=False,
-        ):
+        '''
+        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,
+
+                # XXX: obviously this will deadlock if an on-going stream is
+                # being procesed.
+                # wait_on_aio_task=False,
+            ):
+                raise err
 
     async def wait_asyncio_complete(self) -> None:
         await self._aio_task_complete.wait()
 
-    # def cancel_asyncio_task(self) -> None:
-    #     self._aio_task.cancel()
+    def cancel_asyncio_task(
+        self,
+        msg: str = '',
+    ) -> None:
+        self._aio_task.cancel(
+            msg=msg,
+        )
 
     async def send(self, item: Any) -> None:
         '''
@@ -154,7 +174,6 @@ class LinkedTaskChannel(trio.abc.Channel):
 
 
 def _run_asyncio_task(
-
     func: Callable,
     *,
     qsize: int = 1,
@@ -164,8 +183,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
@@ -203,23 +223,23 @@ def _run_asyncio_task(
     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=cancel_scope,
+        _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 +262,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".
@@ -259,12 +281,21 @@ def _run_asyncio_task(
                 # a ``trio.EndOfChannel`` to the trio (consumer) side.
                 to_trio.close()
 
+            # import pdbp; pdbp.set_trace()
             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,6 +319,10 @@ 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:
@@ -303,13 +338,33 @@ def _run_asyncio_task(
         # 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:
+            task_err: BaseException = _aio_err
+
+            # read again AFTER the `asyncio` side errors in case
+            # it was cancelled due to an error from `trio` (or
+            # some other out of band exc).
+            aio_err: BaseException|None = chan._aio_err
+
+            # always true right?
+            assert (
+                type(_aio_err) is type(aio_err)
+            ), (
+                f'`asyncio`-side task errors mismatch?!?\n\n'
+                f'caught: {_aio_err}\n'
+                f'chan._aio_err: {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,17 +381,15 @@ 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:
+            # import pdbp; pdbp.set_trace()
             # XXX: uhh is this true?
             # assert task_err, f'Asyncio task {task.get_name()} discrepancy!?'
 
             # NOTE: currently mem chan closure may act as a form
-            # of error relay (at least in the ``asyncio.CancelledError``
-            # case) since we have no way to directly trigger a ``trio``
+            # 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()
@@ -347,7 +400,7 @@ def _run_asyncio_task(
                 # 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?
@@ -358,43 +411,57 @@ def _run_asyncio_task(
             #     )
             #     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 cancel_scope.cancelled_caught
+                or
+                not cancel_scope.cancel_called
             ):
-                raise aio_err from task_err
+                # import pdbp; pdbp.set_trace()
+                cancel_scope.cancel()
 
-            # 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
+            if task_err:
+                # XXX raise any `asyncio` side error IFF it doesn't
+                # match the one we just caught from the task above!
+                # (that would indicate something weird/very-wrong
+                # going on?)
+                if aio_err is not task_err:
+                    # import pdbp; pdbp.set_trace()
+                    raise aio_err from task_err
 
     task.add_done_callback(cancel_trio)
     return chan
 
 
+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 +469,247 @@ 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:
+        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()
+
+                if (
+                    not chan._trio_err
+                    and
+                    (fut := aio_task._fut_waiter)
+                ):
+                    fut.set_exception(
+                        TrioTaskExited(
+                            f'The peer `asyncio` task is still blocking/running?\n'
+                            f'>>\n'
+                            f'|_{aio_task!r}\n'
+                        )
+                    )
+                else:
+                    # from tractor._state import is_root_process
+                    # if is_root_process():
+                    #     breakpoint()
+                    #     import pdbp; pdbp.set_trace()
+
+                    aio_taskc_warn: str = (
+                        f'\n'
+                        f'MANUALLY Cancelling `asyncio`-task: {aio_task.get_name()}!\n\n'
+                        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_asyncio_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 +721,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 +762,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,
@@ -684,18 +911,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,16 +942,20 @@ 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..'
-        #     )
-        #     await _debug.maybe_init_greenback(
-        #         force_reload=True,
-        #     )
+        # 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):
             log.runtime(
@@ -731,6 +965,7 @@ def run_as_asyncio_guest(
             )
 
             if isinstance(main_outcome, Error):
+                # import pdbp; pdbp.set_trace()
                 error: BaseException = main_outcome.error
 
                 # show an dedicated `asyncio`-side tb from the error
@@ -750,7 +985,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 +1002,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 +1013,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 +1047,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 +1136,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()