Merge pull request #385 from goodboy/repl_fixture

Add a `repl_fixture` for debugger-interaction-hooks and a newly re-factored subsys pkg: `tractor.devx.debug`

.github/workflows/ci.yml

@@ -8,46 +8,70 @@ on:
   workflow_dispatch:
 
 jobs:
-
+  # ------ sdist ------
-  mypy:
-    name: 'MyPy'
-    runs-on: ubuntu-latest
-
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v2
-
-      - name: Setup python
-        uses: actions/setup-python@v2
-        with:
-          python-version: '3.11'
-
-      - name: Install dependencies
-        run: pip install -U . --upgrade-strategy eager -r requirements-test.txt
-
-      - name: Run MyPy check
-        run: mypy tractor/ --ignore-missing-imports --show-traceback
-
   # test that we can generate a software distribution and install it
   # thus avoid missing file issues after packaging.
+  #
+  # -[x] produce sdist with uv
+  # ------ - ------
   sdist-linux:
     name: 'sdist'
     runs-on: ubuntu-latest
 
     steps:
       - name: Checkout
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
 
-      - name: Setup python
+      - name: Install latest uv
-        uses: actions/setup-python@v2
+        uses: astral-sh/setup-uv@v6
-        with:
-          python-version: '3.11'
 
-      - name: Build sdist
+      - name: Build sdist as tar.gz
-        run: python setup.py sdist --formats=zip
+        run: uv build --sdist --python=3.13
 
-      - name: Install sdist from .zips
+      - name: Install sdist from .tar.gz
-        run: python -m pip install dist/*.zip
+        run: python -m pip install dist/*.tar.gz
+
+  # ------ type-check ------
+  # mypy:
+  #   name: 'MyPy'
+  #   runs-on: ubuntu-latest
+
+  #   steps:
+  #     - name: Checkout
+  #       uses: actions/checkout@v4
+
+  #     - name: Install latest uv
+  #       uses: astral-sh/setup-uv@v6
+
+  #     # faster due to server caching?
+  #     # https://docs.astral.sh/uv/guides/integration/github/#setting-up-python
+  #     - name: "Set up Python"
+  #       uses: actions/setup-python@v6
+  #       with:
+  #         python-version-file: "pyproject.toml"
+
+  #     # w uv
+  #     # - name: Set up Python
+  #     #   run: uv python install
+
+  #     - name: Setup uv venv
+  #       run: uv venv .venv --python=3.13
+
+  #     - name: Install
+  #       run: uv sync --dev
+
+  #     # TODO, ty cmd over repo
+  #     # - name: type check with ty
+  #     #   run: ty ./tractor/
+
+  #     # - uses: actions/cache@v3
+  #     #     name: Cache uv virtenv as default .venv
+  #     #     with:
+  #     #       path: ./.venv
+  #     #       key: venv-${{ hashFiles('uv.lock') }}
+
+  #     - name: Run MyPy check
+  #       run: mypy tractor/ --ignore-missing-imports --show-traceback
 
 
   testing-linux:
@@ -59,32 +83,45 @@ jobs:
       fail-fast: false
       matrix:
         os: [ubuntu-latest]
-        python: ['3.11']
+        python-version: ['3.13']
         spawn_backend: [
           'trio',
-          'mp_spawn',
+          # 'mp_spawn',
-          'mp_forkserver',
+          # 'mp_forkserver',
         ]
 
     steps:
 
-      - name: Checkout
+      - uses: actions/checkout@v4
-        uses: actions/checkout@v2
 
-      - name: Setup python
+      - name: 'Install uv + py-${{ matrix.python-version }}'
-        uses: actions/setup-python@v2
+        uses: astral-sh/setup-uv@v6
         with:
-          python-version: '${{ matrix.python }}'
+          python-version: ${{ matrix.python-version }}
 
-      - name: Install dependencies
+      # GH way.. faster?
-        run: pip install -U . -r requirements-test.txt -r requirements-docs.txt --upgrade-strategy eager
+      # - name: setup-python@v6
+      #   uses: actions/setup-python@v6
+      #   with:
+      #     python-version: '${{ matrix.python-version }}'
 
-      - name: List dependencies
+      # consider caching for speedups?
-        run: pip list
+      # https://docs.astral.sh/uv/guides/integration/github/#caching
+
+      - name: Install the project w uv
+        run: uv sync --all-extras --dev
+
+      # - name: Install dependencies
+      #   run: pip install -U . -r requirements-test.txt -r requirements-docs.txt --upgrade-strategy eager
+
+      - name: List deps tree
+        run: uv tree
 
       - name: Run tests
-        run: pytest tests/ --spawn-backend=${{ matrix.spawn_backend }} -rsx
+        run: uv run pytest tests/ --spawn-backend=${{ matrix.spawn_backend }} -rsx
 
+  # XXX legacy NOTE XXX
+  #
   # We skip 3.10 on windows for now due to not having any collabs to
   # debug the CI failures. Anyone wanting to hack and solve them is very
   # welcome, but our primary user base is not using that OS.

docs/README.rst

@@ -1,8 +1,5 @@
 |logo| ``tractor``: distributed structurred concurrency
 
-|gh_actions|
-|docs|
-
 ``tractor`` is a `structured concurrency`_ (SC), multi-processing_ runtime built on trio_.
 
 Fundamentally, ``tractor`` provides parallelism via
@@ -66,6 +63,13 @@ Features
   - (WIP) a ``TaskMngr``: one-cancels-one style nursery supervisor.
 
 
+Status of `main` / infra
+------------------------
+
+- |gh_actions|
+- |docs|
+
+
 Install
 -------
 ``tractor`` is still in a *alpha-near-beta-stage* for many
@@ -689,9 +693,11 @@ channel`_!
 .. _msgspec: https://jcristharif.com/msgspec/
 .. _guest: https://trio.readthedocs.io/en/stable/reference-lowlevel.html?highlight=guest%20mode#using-guest-mode-to-run-trio-on-top-of-other-event-loops
 
-
+..
-.. |gh_actions| image:: https://img.shields.io/endpoint.svg?url=https%3A%2F%2Factions-badge.atrox.dev%2Fgoodboy%2Ftractor%2Fbadge&style=popout-square
+   NOTE, on generating badge links from the UI
-    :target: https://actions-badge.atrox.dev/goodboy/tractor/goto
+   https://docs.github.com/en/actions/how-tos/monitoring-and-troubleshooting-workflows/monitoring-workflows/adding-a-workflow-status-badge?ref=gitguardian-blog-automated-secrets-detection#using-the-ui
+.. |gh_actions| image:: https://github.com/goodboy/tractor/actions/workflows/ci.yml/badge.svg?branch=main
+    :target: https://github.com/goodboy/tractor/actions/workflows/ci.yml
 
 .. |docs| image:: https://readthedocs.org/projects/tractor/badge/?version=latest
     :target: https://tractor.readthedocs.io/en/latest/?badge=latest

pyproject.toml

@@ -61,7 +61,9 @@ dev = [
   # `tractor.devx` tooling
   "greenback>=1.2.1,<2",
   "stackscope>=0.2.2,<0.3",
-  "typing-extensions>=4.13.2",  # needed for stackscope
+  # ^ requires this?
+  "typing-extensions>=4.14.1",
+
   "pyperclip>=1.9.0",
   "prompt-toolkit>=3.0.50",
   "xonsh>=0.19.2",

tests/devx/test_debugger.py

@@ -317,6 +317,7 @@ def test_subactor_breakpoint(
 
     assert in_prompt_msg(
         child, [
+        'MessagingError:',
         'RemoteActorError:',
          "('breakpoint_forever'",
          'bdb.BdbQuit',

tests/devx/test_tooling.py

@@ -14,6 +14,9 @@ TODO:
 
 '''
 from __future__ import annotations
+from contextlib import (
+    contextmanager as cm,
+)
 import os
 import signal
 import time
@@ -28,6 +31,8 @@ from .conftest import (
     PROMPT,
     _pause_msg,
 )
+
+import pytest
 from pexpect.exceptions import (
     # TIMEOUT,
     EOF,
@@ -116,11 +121,9 @@ def test_shield_pause(
         child.pid,
         signal.SIGINT,
     )
-    from tractor._supervise import _shutdown_msg
     expect(
         child,
-        # 'Shutting down actor runtime',
+        'Shutting down actor runtime',
-        _shutdown_msg,
         timeout=6,
     )
     assert_before(
@@ -185,3 +188,117 @@ def test_breakpoint_hook_restored(
     )
     child.sendline('c')
     child.expect(EOF)
+
+
+
+_to_raise = Exception('Triggering a crash')
+
+
+@pytest.mark.parametrize(
+    'to_raise',
+    [
+        None,
+        _to_raise,
+        RuntimeError('Never crash handle this!'),
+    ],
+)
+@pytest.mark.parametrize(
+    'raise_on_exit',
+    [
+        True,
+        [type(_to_raise)],
+        False,
+    ]
+)
+def test_crash_handler_cms(
+    debug_mode: bool,
+    to_raise: Exception,
+    raise_on_exit: bool|list[Exception],
+):
+    '''
+    Verify the `.devx.open_crash_handler()` API(s) by also
+    (conveniently enough) tesing its `repl_fixture: ContextManager`
+    param support which for this suite allows use to avoid use of
+    a `pexpect`-style-test since we use the fixture to avoid actually
+    entering `PdbpREPL.iteract()` :smirk:
+
+    '''
+    import tractor
+    # import trio
+
+    # state flags
+    repl_acquired: bool = False
+    repl_released: bool = False
+
+    @cm
+    def block_repl_ux(
+        repl: tractor.devx.debug.PdbREPL,
+        maybe_bxerr: (
+            tractor.devx._debug.BoxedMaybeException
+            |None
+        ) = None,
+        enter_repl: bool = True,
+
+    ) -> bool:
+        '''
+        Set pre/post-REPL state vars and bypass actual conole
+        interaction.
+
+        '''
+        nonlocal repl_acquired, repl_released
+
+        # task: trio.Task = trio.lowlevel.current_task()
+        # print(f'pre-REPL active_task={task.name}')
+
+        print('pre-REPL')
+        repl_acquired = True
+        yield False  # never actually .interact()
+        print('post-REPL')
+        repl_released = True
+
+    try:
+        # TODO, with runtime's `debug_mode` setting
+        # -[ ] need to open runtime tho obvi..
+        #
+        # with tractor.devx.maybe_open_crash_handler(
+        #     pdb=True,
+
+        with tractor.devx.open_crash_handler(
+            raise_on_exit=raise_on_exit,
+            repl_fixture=block_repl_ux
+        ) as bxerr:
+            if to_raise is not None:
+                raise to_raise
+
+    except Exception as _exc:
+        exc = _exc
+        if (
+            raise_on_exit is True
+            or
+            type(to_raise) in raise_on_exit
+        ):
+            assert (
+                exc
+                is
+                to_raise
+                is
+                bxerr.value
+            )
+
+        else:
+            raise
+    else:
+        assert (
+            to_raise is None
+            or
+            not raise_on_exit
+            or
+            type(to_raise) not in raise_on_exit
+        )
+        assert bxerr.value is to_raise
+
+    assert bxerr.raise_on_exit == raise_on_exit
+
+    if to_raise is not None:
+        assert repl_acquired
+        assert repl_released

tests/test_clustering.py

@@ -25,17 +25,14 @@ def test_empty_mngrs_input_raises() -> None:
                 ) as portals,
 
                 gather_contexts(
+                    # NOTE: it's the use of inline-generator syntax
+                    # here that causes the empty input.
                     mngrs=(
                         p.open_context(worker) for p in portals.values()
                     ),
-                  # ^^NOTE XXX ^^^
-                    # it's the use of inline-generator syntax here
-                    # that causes the "empty input" -> ValueError,
-                    # see `._clustering` impl.
                 ),
             ):
-                # test should fail if we mk it here!
+                assert 0
-                assert 0, 'Should have raised val-err !?'
 
     with pytest.raises(ValueError):
         trio.run(main)

tests/test_context_stream_semantics.py

@@ -252,7 +252,7 @@ def test_simple_context(
             pass
         except BaseExceptionGroup as beg:
             # XXX: on windows it seems we may have to expect the group error
-            from tractor.trionics import is_multi_cancelled
+            from tractor._exceptions import is_multi_cancelled
             assert is_multi_cancelled(beg)
     else:
         trio.run(main)

tests/test_inter_peer_cancellation.py

@@ -410,6 +410,7 @@ def test_peer_canceller(
     '''
     async def main():
         async with tractor.open_nursery(
+            # NOTE: to halt the peer tasks on ctxc, uncomment this.
             debug_mode=debug_mode,
         ) as an:
             canceller: Portal = await an.start_actor(

tests/test_remote_exc_relay.py

@@ -1,237 +0,0 @@
-'''
-Special case testing for issues not (dis)covered in the primary
-`Context` related functional/scenario suites.
-
-**NOTE: this mod is a WIP** space for handling
-odd/rare/undiscovered/not-yet-revealed faults which either
-loudly (ideal case) breakl our supervision protocol
-or (worst case) result in distributed sys hangs.
-
-Suites here further try to clarify (if [partially] ill-defined) and
-verify our edge case semantics for inter-actor-relayed-exceptions
-including,
-
-- lowlevel: what remote obj-data is interchanged for IPC and what is
-  native-obj form is expected from unpacking in the the new
-  mem-domain.
-
-- which kinds of `RemoteActorError` (and its derivs) are expected by which
-  (types of) peers (parent, child, sibling, etc) with what
-  particular meta-data set such as,
-
-  - `.src_uid`: the original (maybe) peer who raised.
-  - `.relay_uid`: the next-hop-peer who sent it.
-  - `.relay_path`: the sequence of peer actor hops.
-  - `.is_inception`: a predicate that denotes multi-hop remote errors.
-
-- when should `ExceptionGroup`s be relayed from a particular
-  remote endpoint, they should never be caused by implicit `._rpc`
-  nursery machinery!
-
-- various special `trio` edge cases around its cancellation semantics
-  and how we (currently) leverage `trio.Cancelled` as a signal for
-  whether a `Context` task should raise `ContextCancelled` (ctx).
-
-'''
-import pytest
-import trio
-import tractor
-from tractor import (  # typing
-    ActorNursery,
-    Portal,
-    Context,
-    ContextCancelled,
-)
-
-
-@tractor.context
-async def sleep_n_chkpt_in_finally(
-    ctx: Context,
-    sleep_n_raise: bool,
-
-    chld_raise_delay: float,
-    chld_finally_delay: float,
-
-    rent_cancels: bool,
-    rent_ctxc_delay: float,
-
-    expect_exc: str|None = None,
-
-) -> None:
-    '''
-    Sync, open a tn, then wait for cancel, run a chkpt inside
-    the user's `finally:` teardown.
-
-    This covers a footgun case that `trio` core doesn't seem to care about
-    wherein an exc can be masked by a `trio.Cancelled` raised inside a tn emedded
-    `finally:`.
-
-    Also see `test_trioisms::test_acm_embedded_nursery_propagates_enter_err`
-    for the down and gritty details.
-
-    Since a `@context` endpoint fn can also contain code like this,
-    **and** bc we currently have no easy way other then
-    `trio.Cancelled` to signal cancellation on each side of an IPC `Context`,
-    the footgun issue can compound itself as demonstrated in this suite..
-
-    Here are some edge cases codified with our WIP "sclang" syntax
-    (note the parent(rent)/child(chld) naming here is just
-    pragmatism, generally these most of these cases can occurr
-    regardless of the distributed-task's supervision hiearchy),
-
-    - rent c)=> chld.raises-then-taskc-in-finally
-     |_ chld's body raises an `exc: BaseException`.
-      _ in its `finally:` block it runs a chkpoint
-        which raises a taskc (`trio.Cancelled`) which
-        masks `exc` instead raising taskc up to the first tn.
-      _ the embedded/chld tn captures the masking taskc and then
-        raises it up to the ._rpc-ep-tn instead of `exc`.
-      _ the rent thinks the child ctxc-ed instead of errored..
-
-    '''
-    await ctx.started()
-
-    if expect_exc:
-        expect_exc: BaseException = tractor._exceptions.get_err_type(
-            type_name=expect_exc,
-        )
-
-    berr: BaseException|None = None
-    try:
-        if not sleep_n_raise:
-            await trio.sleep_forever()
-        elif sleep_n_raise:
-
-            # XXX this sleep is less then the sleep the parent
-            # does before calling `ctx.cancel()`
-            await trio.sleep(chld_raise_delay)
-
-            # XXX this will be masked by a taskc raised in
-            # the `finally:` if this fn doesn't terminate
-            # before any ctxc-req arrives AND a checkpoint is hit
-            # in that `finally:`.
-            raise RuntimeError('my app krurshed..')
-
-    except BaseException as _berr:
-        berr = _berr
-
-        # TODO: it'd sure be nice to be able to inject our own
-        # `ContextCancelled` here instead of of `trio.Cancelled`
-        # so that our runtime can expect it and this "user code"
-        # would be able to tell the diff between a generic trio
-        # cancel and a tractor runtime-IPC cancel.
-        if expect_exc:
-            if not isinstance(
-                berr,
-                expect_exc,
-            ):
-                raise ValueError(
-                    f'Unexpected exc type ??\n'
-                    f'{berr!r}\n'
-                    f'\n'
-                    f'Expected a {expect_exc!r}\n'
-                )
-
-        raise berr
-
-    # simulate what user code might try even though
-    # it's a known boo-boo..
-    finally:
-        # maybe wait for rent ctxc to arrive
-        with trio.CancelScope(shield=True):
-            await trio.sleep(chld_finally_delay)
-
-        # !!XXX this will raise `trio.Cancelled` which
-        # will mask the RTE from above!!!
-        #
-        # YES, it's the same case as our extant
-        # `test_trioisms::test_acm_embedded_nursery_propagates_enter_err`
-        try:
-            await trio.lowlevel.checkpoint()
-        except trio.Cancelled as taskc:
-            if (scope_err := taskc.__context__):
-                print(
-                    f'XXX MASKED REMOTE ERROR XXX\n'
-                    f'ENDPOINT exception -> {scope_err!r}\n'
-                    f'will be masked by -> {taskc!r}\n'
-                )
-                # await tractor.pause(shield=True)
-
-            raise taskc
-
-
-@pytest.mark.parametrize(
-    'chld_callspec',
-    [
-        dict(
-            sleep_n_raise=None,
-            chld_raise_delay=0.1,
-            chld_finally_delay=0.1,
-            expect_exc='Cancelled',
-            rent_cancels=True,
-            rent_ctxc_delay=0.1,
-        ),
-        dict(
-            sleep_n_raise='RuntimeError',
-            chld_raise_delay=0.1,
-            chld_finally_delay=1,
-            expect_exc='RuntimeError',
-            rent_cancels=False,
-            rent_ctxc_delay=0.1,
-        ),
-    ],
-    ids=lambda item: f'chld_callspec={item!r}'
-)
-def test_unmasked_remote_exc(
-    debug_mode: bool,
-    chld_callspec: dict,
-    tpt_proto: str,
-):
-    expect_exc_str: str|None = chld_callspec['sleep_n_raise']
-    rent_ctxc_delay: float|None = chld_callspec['rent_ctxc_delay']
-    async def main():
-        an: ActorNursery
-        async with tractor.open_nursery(
-            debug_mode=debug_mode,
-            enable_transports=[tpt_proto],
-        ) as an:
-            ptl: Portal = await an.start_actor(
-                'cancellee',
-                enable_modules=[__name__],
-            )
-            ctx: Context
-            async with (
-                ptl.open_context(
-                    sleep_n_chkpt_in_finally,
-                    **chld_callspec,
-                ) as (ctx, sent),
-            ):
-                assert not sent
-                await trio.sleep(rent_ctxc_delay)
-                await ctx.cancel()
-
-                # recv error or result from chld
-                ctxc: ContextCancelled = await ctx.wait_for_result()
-                assert (
-                    ctxc is ctx.outcome
-                    and
-                    isinstance(ctxc, ContextCancelled)
-                )
-
-            # always graceful terminate the sub in non-error cases
-            await an.cancel()
-
-    if expect_exc_str:
-        expect_exc: BaseException = tractor._exceptions.get_err_type(
-            type_name=expect_exc_str,
-        )
-        with pytest.raises(
-            expected_exception=tractor.RemoteActorError,
-        ) as excinfo:
-            trio.run(main)
-
-        rae = excinfo.value
-        assert expect_exc == rae.boxed_type
-
-    else:
-        trio.run(main)

tests/test_trioisms.py

@@ -112,11 +112,55 @@ def test_acm_embedded_nursery_propagates_enter_err(
     '''
     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,
-            tractor.trionics.maybe_raise_from_masking_exc(
+            maybe_raise_from_masking_exc(
                 tn=tn,
                 unmask_from=(
                     trio.Cancelled
@@ -158,60 +202,3 @@ def test_acm_embedded_nursery_propagates_enter_err(
     assert_eg, rest_eg = eg.split(AssertionError)
 
     assert len(assert_eg.exceptions) == 1
-
-
-
-def test_gatherctxs_with_memchan_breaks_multicancelled(
-    debug_mode: bool,
-):
-    '''
-    Demo how a using an `async with sndchan` inside a `.trionics.gather_contexts()` task
-    will break a strict-eg-tn's multi-cancelled absorption..
-
-    '''
-    from tractor import (
-        trionics,
-    )
-
-    @acm
-    async def open_memchan() -> trio.abc.ReceiveChannel:
-
-        task: trio.Task = trio.lowlevel.current_task()
-        print(
-            f'Opening {task!r}\n'
-        )
-
-        # 1 to force eager sending
-        send, recv = trio.open_memory_channel(16)
-
-        try:
-            async with send:
-                yield recv
-        finally:
-            print(
-                f'Closed {task!r}\n'
-            )
-
-
-    async def main():
-        async with (
-            # XXX should ensure ONLY the KBI
-            # is relayed upward
-            trionics.collapse_eg(),
-            trio.open_nursery(
-                # strict_exception_groups=False,
-            ), # as tn,
-
-            trionics.gather_contexts([
-                open_memchan(),
-                open_memchan(),
-            ]) as recv_chans,
-        ):
-            assert len(recv_chans) == 2
-
-            await trio.sleep(1)
-            raise KeyboardInterrupt
-            # tn.cancel_scope.cancel()
-
-    with pytest.raises(KeyboardInterrupt):
-        trio.run(main)

tractor/_clustering.py

@@ -55,17 +55,10 @@ async def open_actor_cluster(
         raise ValueError(
             'Number of names is {len(names)} but count it {count}')
 
-    async with (
+    async with tractor.open_nursery(
-        # tractor.trionics.collapse_eg(),
+        **runtime_kwargs,
-        tractor.open_nursery(
+    ) as an:
-            **runtime_kwargs,
+        async with trio.open_nursery() as n:
-        ) as an
-    ):
-        async with (
-            tractor.trionics.collapse_eg(),
-            trio.open_nursery() as tn,
-            tractor.trionics.maybe_raise_from_masking_exc()
-        ):
             uid = tractor.current_actor().uid
 
             async def _start(name: str) -> None:
@@ -76,8 +69,9 @@ async def open_actor_cluster(
                 )
 
             for name in names:
-                tn.start_soon(_start, name)
+                n.start_soon(_start, name)
 
         assert len(portals) == count
         yield portals
+
         await an.cancel(hard_kill=hard_kill)

tractor/_context.py

@@ -101,9 +101,6 @@ from ._state import (
     debug_mode,
     _ctxvar_Context,
 )
-from .trionics import (
-    collapse_eg,
-)
 # ------ - ------
 if TYPE_CHECKING:
     from ._portal import Portal
@@ -743,8 +740,6 @@ class Context:
             # cancelled, NOT their reported canceller. IOW in the
             # latter case we're cancelled by someone else getting
             # cancelled.
-            #
-            # !TODO, switching to `Actor.aid` here!
             if (canc := error.canceller) == self._actor.uid:
                 whom: str = 'us'
                 self._canceller = canc
@@ -945,7 +940,7 @@ class Context:
         self.cancel_called = True
 
         header: str = (
-            f'Cancelling ctx from {side!r}-side\n'
+            f'Cancelling ctx from {side.upper()}-side\n'
         )
         reminfo: str = (
             # ' =>\n'
@@ -953,7 +948,7 @@ class Context:
             f'\n'
             f'c)=> {self.chan.uid}\n'
             f'   |_[{self.dst_maddr}\n'
-            f'     >> {self.repr_rpc}\n'
+            f'     >>{self.repr_rpc}\n'
             # f'    >> {self._nsf}() -> {codec}[dict]:\n\n'
             # TODO: pull msg-type from spec re #320
         )
@@ -2028,8 +2023,10 @@ async def open_context_from_portal(
     ctxc_from_callee: ContextCancelled|None = None
     try:
         async with (
-            collapse_eg(),
+            trio.open_nursery(
-            trio.open_nursery() as tn,
+                strict_exception_groups=False,
+            ) as tn,
+
             msgops.maybe_limit_plds(
                 ctx=ctx,
                 spec=ctx_meta.get('pld_spec'),

tractor/_discovery.py

@@ -28,10 +28,7 @@ from typing import (
 from contextlib import asynccontextmanager as acm
 
 from tractor.log import get_logger
-from .trionics import (
+from .trionics import gather_contexts
-    gather_contexts,
-    collapse_eg,
-)
 from .ipc import _connect_chan, Channel
 from ._addr import (
     UnwrappedAddress,
@@ -51,7 +48,6 @@ from ._state import (
 
 if TYPE_CHECKING:
     from ._runtime import Actor
-    from .ipc._server import IPCServer
 
 
 log = get_logger(__name__)
@@ -83,7 +79,7 @@ async def get_registry(
         )
     else:
         # TODO: try to look pre-existing connection from
-        # `IPCServer._peers` and use it instead?
+        # `Server._peers` and use it instead?
         async with (
             _connect_chan(addr) as chan,
             open_portal(chan) as regstr_ptl,
@@ -91,6 +87,7 @@ async def get_registry(
             yield regstr_ptl
 
 
+
 @acm
 async def get_root(
     **kwargs,
@@ -114,18 +111,23 @@ def get_peer_by_name(
 ) -> list[Channel]|None:  # at least 1
     '''
     Scan for an existing connection (set) to a named actor
-    and return any channels from `IPCServer._peers: dict`.
+    and return any channels from `Server._peers: dict`.
 
     This is an optimization method over querying the registrar for
     the same info.
 
     '''
     actor: Actor = current_actor()
-    server: IPCServer = actor.ipc_server
+    to_scan: dict[tuple, list[Channel]] = actor.ipc_server._peers.copy()
-    to_scan: dict[tuple, list[Channel]] = server._peers.copy()
+
-    pchan: Channel|None = actor._parent_chan
+    # TODO: is this ever needed? creates a duplicate channel on actor._peers
-    if pchan:
+    # when multiple find_actor calls are made to same actor from a single ctx
-        to_scan[pchan.uid].append(pchan)
+    # which causes actor exit to hang waiting forever on
+    # `actor._no_more_peers.wait()` in `_runtime.async_main`
+
+    # pchan: Channel|None = actor._parent_chan
+    # if pchan and pchan.uid not in to_scan:
+    #     to_scan[pchan.uid].append(pchan)
 
     for aid, chans in to_scan.items():
         _, peer_name = aid
@@ -251,12 +253,9 @@ async def find_actor(
         for addr in registry_addrs
     )
     portals: list[Portal]
-    async with (
+    async with gather_contexts(
-        collapse_eg(),
+        mngrs=maybe_portals,
-        gather_contexts(
+    ) as portals:
-            mngrs=maybe_portals,
-        ) as portals,
-    ):
         # log.runtime(
         #     'Gathered portals:\n'
         #     f'{portals}'

tractor/_entry.py

@@ -21,7 +21,7 @@ Sub-process entry points.
 from __future__ import annotations
 from functools import partial
 import multiprocessing as mp
-# import os
+import os
 from typing import (
     Any,
     TYPE_CHECKING,
@@ -38,7 +38,6 @@ from .devx import (
     _frame_stack,
     pformat,
 )
-# from .msg import pretty_struct
 from .to_asyncio import run_as_asyncio_guest
 from ._addr import UnwrappedAddress
 from ._runtime import (
@@ -128,13 +127,20 @@ def _trio_main(
 
     if actor.loglevel is not None:
         get_console_log(actor.loglevel)
+        actor_info: str = (
+            f'|_{actor}\n'
+            f'  uid: {actor.uid}\n'
+            f'  pid: {os.getpid()}\n'
+            f'  parent_addr: {parent_addr}\n'
+            f'  loglevel: {actor.loglevel}\n'
+        )
         log.info(
-            f'Starting `trio` subactor from parent @ '
+            'Starting new `trio` subactor\n'
-            f'{parent_addr}\n'
             +
             pformat.nest_from_op(
                 input_op='>(',  # see syntax ideas above
-                text=f'{actor}',
+                text=actor_info,
+                nest_indent=2,  # since "complete"
             )
         )
     logmeth = log.info
@@ -143,7 +149,7 @@ def _trio_main(
         +
         pformat.nest_from_op(
             input_op=')>',  # like a "closed-to-play"-icon from super perspective
-            text=f'{actor}',
+            text=actor_info,
             nest_indent=1,
         )
     )
@@ -161,7 +167,7 @@ def _trio_main(
             +
             pformat.nest_from_op(
                 input_op='c)>',  # closed due to cancel (see above)
-                text=f'{actor}',
+                text=actor_info,
             )
         )
     except BaseException as err:
@@ -171,7 +177,7 @@ def _trio_main(
             +
             pformat.nest_from_op(
                 input_op='x)>',  # closed by error
-                text=f'{actor}',
+                text=actor_info,
             )
         )
         # NOTE since we raise a tb will already be shown on the

tractor/_exceptions.py

@@ -1246,6 +1246,55 @@ def unpack_error(
     return exc
 
 
+def is_multi_cancelled(
+    exc: BaseException|BaseExceptionGroup,
+
+    ignore_nested: set[BaseException] = set(),
+
+) -> bool|BaseExceptionGroup:
+    '''
+    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.update({trio.Cancelled})
+
+    if isinstance(exc, BaseExceptionGroup):
+        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
+
+
 def _raise_from_unexpected_msg(
     ctx: Context,
     msg: MsgType,

tractor/_portal.py

@@ -39,10 +39,7 @@ import warnings
 
 import trio
 
-from .trionics import (
+from .trionics import maybe_open_nursery
-    maybe_open_nursery,
-    collapse_eg,
-)
 from ._state import (
     current_actor,
 )
@@ -118,10 +115,6 @@ class Portal:
 
     @property
     def chan(self) -> Channel:
-        '''
-        Ref to this ctx's underlying `tractor.ipc.Channel`.
-
-        '''
         return self._chan
 
     @property
@@ -181,17 +174,10 @@ class Portal:
 
         # not expecting a "main" result
         if self._expect_result_ctx is None:
-            peer_id: str = f'{self.channel.aid.reprol()!r}'
             log.warning(
-                f'Portal to peer {peer_id} will not deliver a final result?\n'
+                f"Portal for {self.channel.aid} not expecting a final"
-                f'\n'
+                " result?\nresult() should only be called if subactor"
-                f'Context.result() can only be called by the parent of '
+                " was spawned with `ActorNursery.run_in_actor()`")
-                f'a sub-actor when it was spawned with '
-                f'`ActorNursery.run_in_actor()`'
-                f'\n'
-                f'Further this `ActorNursery`-method-API will deprecated in the'
-                f'near fututre!\n'
-            )
             return NoResult
 
         # expecting a "main" result
@@ -224,7 +210,6 @@ class Portal:
         typname: str = type(self).__name__
         log.warning(
             f'`{typname}.result()` is DEPRECATED!\n'
-            f'\n'
             f'Use `{typname}.wait_for_result()` instead!\n'
         )
         return await self.wait_for_result(
@@ -236,10 +221,8 @@ class Portal:
         # terminate all locally running async generator
         # IPC calls
         if self._streams:
-            peer_id: str = f'{self.channel.aid.reprol()!r}'
+            log.cancel(
-            report: str = (
+                f"Cancelling all streams with {self.channel.aid}")
-                f'Cancelling all msg-streams with {peer_id}\n'
-            )
             for stream in self._streams.copy():
                 try:
                     await stream.aclose()
@@ -248,18 +231,10 @@ class Portal:
                     # (unless of course at some point down the road we
                     # won't expect this to always be the case or need to
                     # detect it for respawning purposes?)
-                    report += (
+                    log.debug(f"{stream} was already closed.")
-                        f'->) {stream!r} already closed\n'
-                    )
-
-            log.cancel(report)
 
     async def aclose(self):
-        log.debug(
+        log.debug(f"Closing {self}")
-            f'Closing portal\n'
-            f'>}}\n'
-            f'|_{self}\n'
-        )
         # TODO: once we move to implementing our own `ReceiveChannel`
         # (including remote task cancellation inside its `.aclose()`)
         # we'll need to .aclose all those channels here
@@ -285,18 +260,19 @@ class Portal:
         __runtimeframe__: int = 1  # noqa
 
         chan: Channel = self.channel
-        peer_id: str = f'{self.channel.aid.reprol()!r}'
         if not chan.connected():
             log.runtime(
-                'Peer {peer_id} is already disconnected\n'
+                'This channel is already closed, skipping cancel request..'
-                '-> skipping cancel request..\n'
             )
             return False
 
+        reminfo: str = (
+            f'c)=> {self.channel.aid}\n'
+            f'  |_{chan}\n'
+        )
         log.cancel(
-            f'Sending actor-runtime-cancel-req to peer\n'
+            f'Requesting actor-runtime cancel for peer\n\n'
-            f'\n'
+            f'{reminfo}'
-            f'c)=> {peer_id}\n'
         )
 
         # XXX the one spot we set it?
@@ -321,9 +297,8 @@ class Portal:
                 # may timeout and we never get an ack (obvi racy)
                 # but that doesn't mean it wasn't cancelled.
                 log.debug(
-                    f'May have failed to cancel peer?\n'
+                    'May have failed to cancel peer?\n'
-                    f'\n'
+                    f'{reminfo}'
-                    f'c)=?> {peer_id}\n'
                 )
 
             # if we get here some weird cancellation case happened
@@ -341,22 +316,22 @@ class Portal:
 
             TransportClosed,
         ) as tpt_err:
-            ipc_borked_report: str = (
+            report: str = (
-                f'IPC for actor already closed/broken?\n\n'
+                f'IPC chan for actor already closed or broken?\n\n'
-                f'\n'
+                f'{self.channel.aid}\n'
-                f'c)=x> {peer_id}\n'
+                f' |_{self.channel}\n'
             )
             match tpt_err:
                 case TransportClosed():
-                    log.debug(ipc_borked_report)
+                    log.debug(report)
                 case _:
-                    ipc_borked_report += (
+                    report += (
                         f'\n'
                         f'Unhandled low-level transport-closed/error during\n'
                         f'Portal.cancel_actor()` request?\n'
                         f'<{type(tpt_err).__name__}( {tpt_err} )>\n'
                     )
-                    log.warning(ipc_borked_report)
+                    log.warning(report)
 
             return False
 
@@ -513,13 +488,10 @@ class Portal:
                 with trio.CancelScope(shield=True):
                     await ctx.cancel()
 
-            except trio.ClosedResourceError as cre:
+            except trio.ClosedResourceError:
                 # if the far end terminates before we send a cancel the
                 # underlying transport-channel may already be closed.
-                log.cancel(
+                log.cancel(f'Context {ctx} was already closed?')
-                    f'Context.cancel() -> {cre!r}\n'
-                    f'cid: {ctx.cid!r} already closed?\n'
-                )
 
             # XXX: should this always be done?
             # await recv_chan.aclose()
@@ -586,13 +558,14 @@ async def open_portal(
     assert actor
     was_connected: bool = False
 
-    async with (
+    async with maybe_open_nursery(
-        collapse_eg(),
+        tn,
-        maybe_open_nursery(
+        shield=shield,
-            tn,
+        strict_exception_groups=False,
-            shield=shield,
+        # ^XXX^ TODO? soo roll our own then ??
-        ) as tn,
+        # -> since we kinda want the "if only one `.exception` then
-    ):
+        # just raise that" interface?
+    ) as tn:
 
         if not channel.connected():
             await channel.connect()

tractor/_root.py

@@ -37,11 +37,16 @@ import warnings
 
 import trio
 
-from . import _runtime
+from ._runtime import (
+    Actor,
+    Arbiter,
+    # TODO: rename and make a non-actor subtype?
+    # Arbiter as Registry,
+    async_main,
+)
 from .devx import (
     debug,
     _frame_stack,
-    pformat as _pformat,
 )
 from . import _spawn
 from . import _state
@@ -56,12 +61,9 @@ from ._addr import (
     mk_uuid,
     wrap_address,
 )
-from .trionics import (
-    is_multi_cancelled,
-    collapse_eg,
-)
 from ._exceptions import (
     RuntimeFailure,
+    is_multi_cancelled,
 )
 
 
@@ -97,7 +99,7 @@ async def maybe_block_bp(
     ):
         logger.info(
             f'Found `greenback` installed @ {maybe_mod}\n'
-            f'Enabling `tractor.pause_from_sync()` support!\n'
+            'Enabling `tractor.pause_from_sync()` support!\n'
         )
         os.environ['PYTHONBREAKPOINT'] = (
             'tractor.devx.debug._sync_pause_from_builtin'
@@ -192,19 +194,13 @@ async def open_root_actor(
     # read-only state to sublayers?
     # extra_rt_vars: dict|None = None,
 
-) -> _runtime.Actor:
+) -> Actor:
     '''
-    Initialize the `tractor` runtime by starting a "root actor" in
+    Runtime init entry point for ``tractor``.
-    a parent-most Python process.
-
-    All (disjoint) actor-process-trees-as-programs are created via
-    this entrypoint.
 
     '''
     # XXX NEVER allow nested actor-trees!
-    if already_actor := _state.current_actor(
+    if already_actor := _state.current_actor(err_on_no_runtime=False):
-        err_on_no_runtime=False,
-    ):
         rtvs: dict[str, Any] = _state._runtime_vars
         root_mailbox: list[str, int] = rtvs['_root_mailbox']
         registry_addrs: list[list[str, int]] = rtvs['_registry_addrs']
@@ -274,20 +270,14 @@ async def open_root_actor(
                 DeprecationWarning,
                 stacklevel=2,
             )
-            uw_reg_addrs = [arbiter_addr]
+            registry_addrs = [arbiter_addr]
 
-        uw_reg_addrs = registry_addrs
+        if not registry_addrs:
-        if not uw_reg_addrs:
+            registry_addrs: list[UnwrappedAddress] = default_lo_addrs(
-            uw_reg_addrs: list[UnwrappedAddress] = default_lo_addrs(
                 enable_transports
             )
 
-        # must exist by now since all below code is dependent
+        assert registry_addrs
-        assert uw_reg_addrs
-        registry_addrs: list[Address] = [
-            wrap_address(uw_addr)
-            for uw_addr in uw_reg_addrs
-        ]
 
         loglevel = (
             loglevel
@@ -336,10 +326,10 @@ async def open_root_actor(
             enable_stack_on_sig()
 
         # closed into below ping task-func
-        ponged_addrs: list[Address] = []
+        ponged_addrs: list[UnwrappedAddress] = []
 
         async def ping_tpt_socket(
-            addr: Address,
+            addr: UnwrappedAddress,
             timeout: float = 1,
         ) -> None:
             '''
@@ -359,22 +349,17 @@ async def open_root_actor(
                 # be better to eventually have a "discovery" protocol
                 # with basic handshake instead?
                 with trio.move_on_after(timeout):
-                    async with _connect_chan(addr.unwrap()):
+                    async with _connect_chan(addr):
                         ponged_addrs.append(addr)
 
             except OSError:
-                # ?TODO, make this a "discovery" log level?
+                # TODO: make this a "discovery" log level?
                 logger.info(
-                    f'No root-actor registry found @ {addr!r}\n'
+                    f'No actor registry found @ {addr}\n'
                 )
 
-        # !TODO, this is basically just another (abstract)
-        # happy-eyeballs, so we should try for formalize it somewhere
-        # in a `.[_]discovery` ya?
-        #
         async with trio.open_nursery() as tn:
-            for uw_addr in uw_reg_addrs:
+            for addr in registry_addrs:
-                addr: Address = wrap_address(uw_addr)
                 tn.start_soon(
                     ping_tpt_socket,
                     addr,
@@ -396,35 +381,31 @@ async def open_root_actor(
                 f'Registry(s) seem(s) to exist @ {ponged_addrs}'
             )
 
-            actor = _runtime.Actor(
+            actor = Actor(
                 name=name or 'anonymous',
                 uuid=mk_uuid(),
                 registry_addrs=ponged_addrs,
                 loglevel=loglevel,
                 enable_modules=enable_modules,
             )
-            # **DO NOT** use the registry_addrs as the
+            # DO NOT use the registry_addrs as the transport server
-            # ipc-transport-server's bind-addrs as this is
+            # addrs for this new non-registar, root-actor.
-            # a new NON-registrar, ROOT-actor.
-            #
-            # XXX INSTEAD, bind random addrs using the same tpt
-            # proto.
             for addr in ponged_addrs:
+                waddr: Address = wrap_address(addr)
                 trans_bind_addrs.append(
-                    addr.get_random(
+                    waddr.get_random(bindspace=waddr.bindspace)
-                        bindspace=addr.bindspace,
-                    )
                 )
 
         # Start this local actor as the "registrar", aka a regular
         # actor who manages the local registry of "mailboxes" of
         # other process-tree-local sub-actors.
         else:
+
             # NOTE that if the current actor IS THE REGISTAR, the
             # following init steps are taken:
             # - the tranport layer server is bound to each addr
             #   pair defined in provided registry_addrs, or the default.
-            trans_bind_addrs = uw_reg_addrs
+            trans_bind_addrs = registry_addrs
 
             # - it is normally desirable for any registrar to stay up
             #   indefinitely until either all registered (child/sub)
@@ -435,8 +416,7 @@ async def open_root_actor(
             # https://github.com/goodboy/tractor/pull/348
             # https://github.com/goodboy/tractor/issues/296
 
-            # TODO: rename as `RootActor` or is that even necessary?
+            actor = Arbiter(
-            actor = _runtime.Arbiter(
                 name=name or 'registrar',
                 uuid=mk_uuid(),
                 registry_addrs=registry_addrs,
@@ -448,16 +428,6 @@ async def open_root_actor(
             # `.trio.run()`.
             actor._infected_aio = _state._runtime_vars['_is_infected_aio']
 
-        # NOTE, only set the loopback addr for the
-        # process-tree-global "root" mailbox since all sub-actors
-        # should be able to speak to their root actor over that
-        # channel.
-        raddrs: list[Address] = _state._runtime_vars['_root_addrs']
-        raddrs.extend(trans_bind_addrs)
-        # TODO, remove once we have also removed all usage;
-        # eventually all (root-)registry apis should expect > 1 addr.
-        _state._runtime_vars['_root_mailbox'] = raddrs[0]
-
         # Start up main task set via core actor-runtime nurseries.
         try:
             # assign process-local actor
@@ -465,27 +435,21 @@ async def open_root_actor(
 
             # start local channel-server and fake the portal API
             # NOTE: this won't block since we provide the nursery
-            report: str = f'Starting actor-runtime for {actor.aid.reprol()!r}\n'
+            ml_addrs_str: str = '\n'.join(
-            if reg_addrs := actor.registry_addrs:
+                f'@{addr}' for addr in trans_bind_addrs
-                report += (
+            )
-                    '-> Opening new registry @ '
+            logger.info(
-                    +
+                f'Starting local {actor.uid} on the following transport addrs:\n'
-                    '\n'.join(
+                f'{ml_addrs_str}'
-                        f'{addr}' for addr in reg_addrs
+            )
-                    )
-                )
-            logger.info(f'{report}\n')
 
-            # start runtime in a bg sub-task, yield to caller.
+            # start the actor runtime in a new task
-            async with (
+            async with trio.open_nursery(
-                collapse_eg(),
+                strict_exception_groups=False,
-                trio.open_nursery() as root_tn,
+                # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
+            ) as nursery:
 
-                # XXX, finally-footgun below?
+                # ``_runtime.async_main()`` creates an internal nursery
-                # -> see note on why shielding.
-                # maybe_raise_from_masking_exc(),
-            ):
-                # `_runtime.async_main()` creates an internal nursery
                 # and blocks here until any underlying actor(-process)
                 # tree has terminated thereby conducting so called
                 # "end-to-end" structured concurrency throughout an
@@ -493,9 +457,9 @@ async def open_root_actor(
                 # "actor runtime" primitives are SC-compat and thus all
                 # transitively spawned actors/processes must be as
                 # well.
-                await root_tn.start(
+                await nursery.start(
                     partial(
-                        _runtime.async_main,
+                        async_main,
                         actor,
                         accept_addrs=trans_bind_addrs,
                         parent_addr=None
@@ -543,7 +507,7 @@ async def open_root_actor(
                     raise
 
                 finally:
-                    # NOTE/TODO?, not sure if we'll ever need this but it's
+                    # NOTE: not sure if we'll ever need this but it's
                     # possibly better for even more determinism?
                     # logger.cancel(
                     #     f'Waiting on {len(nurseries)} nurseries in root..')
@@ -552,21 +516,12 @@ async def open_root_actor(
                     #     for an in nurseries:
                     #         tempn.start_soon(an.exited.wait)
 
-                    op_nested_actor_repr: str = _pformat.nest_from_op(
-                        input_op='>) ',
-                        text=actor.pformat(),
-                        nest_prefix='|_',
-                    )
                     logger.info(
                         f'Closing down root actor\n'
-                        f'{op_nested_actor_repr}'
+                        f'>)\n'
+                        f'|_{actor}\n'
                     )
-                    # XXX, THIS IS A *finally-footgun*!
+                    await actor.cancel(None)  # self cancel
-                    # -> though already shields iternally it can
-                    # taskc here and mask underlying errors raised in
-                    # the try-block above?
-                    with trio.CancelScope(shield=True):
-                        await actor.cancel(None)  # self cancel
         finally:
             # revert all process-global runtime state
             if (
@@ -579,16 +534,10 @@ async def open_root_actor(
             _state._current_actor = None
             _state._last_actor_terminated = actor
 
-            sclang_repr: str = _pformat.nest_from_op(
+            logger.runtime(
-                input_op=')>',
-                text=actor.pformat(),
-                nest_prefix='|_',
-                nest_indent=1,
-            )
-
-            logger.info(
                 f'Root actor terminated\n'
-                f'{sclang_repr}'
+                f')>\n'
+                f' |_{actor}\n'
             )
 
 

tractor/_rpc.py

@@ -37,7 +37,6 @@ import warnings
 
 import trio
 from trio import (
-    Cancelled,
     CancelScope,
     Nursery,
     TaskStatus,
@@ -53,18 +52,13 @@ from ._exceptions import (
     ModuleNotExposed,
     MsgTypeError,
     TransportClosed,
+    is_multi_cancelled,
     pack_error,
     unpack_error,
 )
-from .trionics import (
-    collapse_eg,
-    is_multi_cancelled,
-    maybe_raise_from_masking_exc,
-)
 from .devx import (
     debug,
     add_div,
-    pformat as _pformat,
 )
 from . import _state
 from .log import get_logger
@@ -73,7 +67,7 @@ from .msg import (
     MsgCodec,
     PayloadT,
     NamespacePath,
-    pretty_struct,
+    # pretty_struct,
     _ops as msgops,
 )
 from tractor.msg.types import (
@@ -221,18 +215,11 @@ async def _invoke_non_context(
             task_status.started(ctx)
             result = await coro
             fname: str = func.__name__
-
-            op_nested_task: str = _pformat.nest_from_op(
-                input_op=f')> cid: {ctx.cid!r}',
-                text=f'{ctx._task}',
-                nest_indent=1,  # under >
-            )
             log.runtime(
-                f'RPC task complete\n'
+                'RPC complete:\n'
-                f'\n'
+                f'task: {ctx._task}\n'
-                f'{op_nested_task}\n'
+                f'|_cid={ctx.cid}\n'
-                f'\n'
+                f'|_{fname}() -> {pformat(result)}\n'
-                f')> {fname}() -> {pformat(result)}\n'
             )
 
             # NOTE: only send result if we know IPC isn't down
@@ -263,7 +250,7 @@ async def _errors_relayed_via_ipc(
     ctx: Context,
     is_rpc: bool,
 
-    hide_tb: bool = True,
+    hide_tb: bool = False,
     debug_kbis: bool = False,
     task_status: TaskStatus[
         Context | BaseException
@@ -388,9 +375,9 @@ async def _errors_relayed_via_ipc(
     # they can be individually ccancelled.
     finally:
 
-        # if the error is not from user code and instead a failure of
+        # if the error is not from user code and instead a failure
-        # an internal-runtime-RPC or IPC-connection, we do (prolly) want
+        # of a runtime RPC or transport failure we do prolly want to
-        # to show this frame!
+        # show this frame
         if (
             rpc_err
             and (
@@ -629,40 +616,32 @@ async def _invoke(
         #  -> the below scope is never exposed to the
         #     `@context` marked RPC function.
         # - `._portal` is never set.
-        scope_err: BaseException|None = None
         try:
-            # TODO: better `trionics` primitive/tooling usage here!
+            tn: trio.Nursery
-            # -[ ] should would be nice to have our `TaskMngr`
-            #   nursery here!
-            # -[ ] payload value checking like we do with
-            #   `.started()` such that the debbuger can engage
-            #   here in the child task instead of waiting for the
-            #   parent to crash with it's own MTE..
-            #
-            tn: Nursery
             rpc_ctx_cs: CancelScope
             async with (
-                collapse_eg(),
+                trio.open_nursery(
-                trio.open_nursery() as tn,
+                    strict_exception_groups=False,
+                    # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
+
+                ) as tn,
                 msgops.maybe_limit_plds(
                     ctx=ctx,
                     spec=ctx_meta.get('pld_spec'),
                     dec_hook=ctx_meta.get('dec_hook'),
                 ),
-
-                # XXX NOTE, this being the "most embedded"
-                # scope ensures unasking of the `await coro` below
-                # *should* never be interfered with!!
-                maybe_raise_from_masking_exc(
-                    tn=tn,
-                    unmask_from=Cancelled,
-                ) as _mbme,  # maybe boxed masked exc
             ):
                 ctx._scope_nursery = tn
                 rpc_ctx_cs = ctx._scope = tn.cancel_scope
                 task_status.started(ctx)
 
-                # invoke user endpoint fn.
+                # TODO: better `trionics` tooling:
+                # -[ ] should would be nice to have our `TaskMngr`
+                #   nursery here!
+                # -[ ] payload value checking like we do with
+                #   `.started()` such that the debbuger can engage
+                #   here in the child task instead of waiting for the
+                #   parent to crash with it's own MTE..
                 res: Any|PayloadT = await coro
                 return_msg: Return|CancelAck = return_msg_type(
                     cid=cid,
@@ -672,8 +651,7 @@ async def _invoke(
                 ctx._result = res
                 log.runtime(
                     f'Sending result msg and exiting {ctx.side!r}\n'
-                    f'\n'
+                    f'{return_msg}\n'
-                    f'{pretty_struct.pformat(return_msg)}\n'
                 )
                 await chan.send(return_msg)
 
@@ -765,48 +743,39 @@ async def _invoke(
             BaseExceptionGroup,
             BaseException,
             trio.Cancelled,
-        ) as _scope_err:
+
-            scope_err = _scope_err
+        ) as scope_error:
             if (
-                isinstance(scope_err, RuntimeError)
+                isinstance(scope_error, RuntimeError)
-                and
+                and scope_error.args
-                scope_err.args
+                and 'Cancel scope stack corrupted' in scope_error.args[0]
-                and
-                'Cancel scope stack corrupted' in scope_err.args[0]
             ):
                 log.exception('Cancel scope stack corrupted!?\n')
                 # debug.mk_pdb().set_trace()
 
             # always set this (child) side's exception as the
             # local error on the context
-            ctx._local_error: BaseException = scope_err
+            ctx._local_error: BaseException = scope_error
             # ^-TODO-^ question,
             # does this matter other then for
             # consistentcy/testing?
             # |_ no user code should be in this scope at this point
             #    AND we already set this in the block below?
 
-            # XXX if a remote error was set then likely the
+            # if a remote error was set then likely the
-            # exc group was raised due to that, so
+            # exception group was raised due to that, so
             # and we instead raise that error immediately!
-            maybe_re: (
+            ctx.maybe_raise()
-                ContextCancelled|RemoteActorError
-            ) = ctx.maybe_raise()
-            if maybe_re:
-                log.cancel(
-                    f'Suppressing remote-exc from peer,\n'
-                    f'{maybe_re!r}\n'
-                )
 
             # maybe TODO: pack in come kinda
             # `trio.Cancelled.__traceback__` here so they can be
             # unwrapped and displayed on the caller side? no se..
-            raise scope_err
+            raise
 
         # `@context` entrypoint task bookeeping.
         # i.e. only pop the context tracking if used ;)
         finally:
-            assert chan.aid
+            assert chan.uid
 
             # don't pop the local context until we know the
             # associated child isn't in debug any more
@@ -833,19 +802,16 @@ async def _invoke(
                     descr_str += (
                         f'\n{merr!r}\n'  # needed?
                         f'{tb_str}\n'
-                        f'\n'
-                        f'scope_error:\n'
-                        f'{scope_err!r}\n'
                     )
                 else:
                     descr_str += f'\n{merr!r}\n'
             else:
-                descr_str += f'\nwith final result {ctx.outcome!r}\n'
+                descr_str += f'\nand final result {ctx.outcome!r}\n'
 
             logmeth(
-                f'{message}\n'
+                message
-                f'\n'
+                +
-                f'{descr_str}\n'
+                descr_str
             )
 
 
@@ -1012,6 +978,8 @@ async def process_messages(
                         cid=cid,
                         kwargs=kwargs,
                     ):
+                        kwargs |= {'req_chan': chan}
+
                         # XXX NOTE XXX don't start entire actor
                         # runtime cancellation if this actor is
                         # currently in debug mode!
@@ -1030,14 +998,14 @@ async def process_messages(
                                 cid,
                                 chan,
                                 actor.cancel,
-                                kwargs | {'req_chan': chan},
+                                kwargs,
                                 is_rpc=False,
                                 return_msg_type=CancelAck,
                             )
 
                         log.runtime(
-                            'Cancelling RPC-msg-loop with peer\n'
+                            'Cancelling IPC transport msg-loop with peer:\n'
-                            f'->c}} {chan.aid.reprol()}@[{chan.maddr}]\n'
+                            f'|_{chan}\n'
                         )
                         loop_cs.cancel()
                         break
@@ -1050,7 +1018,7 @@ async def process_messages(
                     ):
                         target_cid: str = kwargs['cid']
                         kwargs |= {
-                            'requesting_aid': chan.aid,
+                            'requesting_uid': chan.uid,
                             'ipc_msg': msg,
 
                             # XXX NOTE! ONLY the rpc-task-owning
@@ -1086,34 +1054,21 @@ async def process_messages(
                         ns=ns,
                         func=funcname,
                         kwargs=kwargs,  # type-spec this? see `msg.types`
-                        uid=actor_uuid,
+                        uid=actorid,
                     ):
-                        if actor_uuid != chan.aid.uid:
-                            raise RuntimeError(
-                                f'IPC <Start> msg <-> chan.aid mismatch!?\n'
-                                f'Channel.aid = {chan.aid!r}\n'
-                                f'Start.uid = {actor_uuid!r}\n'
-                            )
-                        # await debug.pause()
-                        op_repr: str = 'Start <=) '
-                        req_repr: str = _pformat.nest_from_op(
-                            input_op=op_repr,
-                            op_suffix='',
-                            nest_prefix='',
-                            text=f'{chan}',
-
-                            nest_indent=len(op_repr)-1,
-                            rm_from_first_ln='<',
-                            # ^XXX, subtract -1 to account for
-                            # <Channel
-                            # ^_chevron to be stripped
-                        )
                         start_status: str = (
-                            'Handling RPC request\n'
+                            'Handling RPC `Start` request\n'
-                            f'{req_repr}\n'
+                            f'<= peer: {actorid}\n\n'
-                            f'\n'
+                            f'  |_{chan}\n'
-                            f'->{{ ipc-context-id: {cid!r}\n'
+                            f'  |_cid: {cid}\n\n'
-                            f'->{{ nsp for fn: `{ns}.{funcname}({kwargs})`\n'
+                            # f'  |_{ns}.{funcname}({kwargs})\n'
+                            f'>> {actor.uid}\n'
+                            f'  |_{actor}\n'
+                            f'   -> nsp: `{ns}.{funcname}({kwargs})`\n'
+
+                            # f'  |_{ns}.{funcname}({kwargs})\n\n'
+
+                            # f'{pretty_struct.pformat(msg)}\n'
                         )
 
                         # runtime-internal endpoint: `Actor.<funcname>`
@@ -1142,6 +1097,10 @@ async def process_messages(
                                 await chan.send(err_msg)
                                 continue
 
+                        start_status += (
+                            f'   -> func: {func}\n'
+                        )
+
                         # schedule a task for the requested RPC function
                         # in the actor's main "service nursery".
                         #
@@ -1149,7 +1108,7 @@ async def process_messages(
                         # supervision isolation? would avoid having to
                         # manage RPC tasks individually in `._rpc_tasks`
                         # table?
-                        start_status += '->( scheduling new task..\n'
+                        start_status += '   -> scheduling new task..\n'
                         log.runtime(start_status)
                         try:
                             ctx: Context = await actor._service_n.start(
@@ -1233,24 +1192,12 @@ async def process_messages(
             # END-OF `async for`:
             # IPC disconnected via `trio.EndOfChannel`, likely
             # due to a (graceful) `Channel.aclose()`.
-
-            chan_op_repr: str = '<=x] '
-            chan_repr: str = _pformat.nest_from_op(
-                input_op=chan_op_repr,
-                op_suffix='',
-                nest_prefix='',
-                text=chan.pformat(),
-                nest_indent=len(chan_op_repr)-1,
-                rm_from_first_ln='<',
-            )
             log.runtime(
-                f'IPC channel disconnected\n'
+                f'channel for {chan.uid} disconnected, cancelling RPC tasks\n'
-                f'{chan_repr}\n'
+                f'|_{chan}\n'
-                f'\n'
-                f'->c) cancelling RPC tasks.\n'
             )
             await actor.cancel_rpc_tasks(
-                req_aid=actor.aid,
+                req_uid=actor.uid,
                 # a "self cancel" in terms of the lifetime of the
                 # IPC connection which is presumed to be the
                 # source of any requests for spawned tasks.
@@ -1322,37 +1269,13 @@ async def process_messages(
     finally:
         # msg debugging for when he machinery is brokey
         if msg is None:
-            message: str = 'Exiting RPC-loop without receiving a msg?'
+            message: str = 'Exiting IPC msg loop without receiving a msg?'
         else:
-            task_op_repr: str = ')>'
-            task: trio.Task = trio.lowlevel.current_task()
-
-            # maybe add cancelled opt prefix
-            if task._cancel_status.effectively_cancelled:
-                task_op_repr = 'c' + task_op_repr
-
-            task_repr: str = _pformat.nest_from_op(
-                input_op=task_op_repr,
-                text=f'{task!r}',
-                nest_indent=1,
-            )
-            # chan_op_repr: str = '<=} '
-            # chan_repr: str = _pformat.nest_from_op(
-            #     input_op=chan_op_repr,
-            #     op_suffix='',
-            #     nest_prefix='',
-            #     text=chan.pformat(),
-            #     nest_indent=len(chan_op_repr)-1,
-            #     rm_from_first_ln='<',
-            # )
             message: str = (
-                f'Exiting RPC-loop with final msg\n'
+                'Exiting IPC msg loop with final msg\n\n'
-                f'\n'
+                f'<= peer: {chan.uid}\n'
-                # f'{chan_repr}\n'
+                f'  |_{chan}\n\n'
-                f'{task_repr}\n'
+                # f'{pretty_struct.pformat(msg)}'
-                f'\n'
-                f'{pretty_struct.pformat(msg)}'
-                f'\n'
             )
 
         log.runtime(message)

tractor/_runtime.py

@@ -55,7 +55,6 @@ from typing import (
     TYPE_CHECKING,
 )
 import uuid
-import textwrap
 from types import ModuleType
 import warnings
 
@@ -74,9 +73,6 @@ from tractor.msg import (
     pretty_struct,
     types as msgtypes,
 )
-from .trionics import (
-    collapse_eg,
-)
 from .ipc import (
     Channel,
     # IPCServer,  # causes cycles atm..
@@ -101,10 +97,7 @@ from ._exceptions import (
     MsgTypeError,
     unpack_error,
 )
-from .devx import (
+from .devx import debug
-    debug,
-    pformat as _pformat
-)
 from ._discovery import get_registry
 from ._portal import Portal
 from . import _state
@@ -213,7 +206,7 @@ class Actor:
         *,
         enable_modules: list[str] = [],
         loglevel: str|None = None,
-        registry_addrs: list[Address]|None = None,
+        registry_addrs: list[UnwrappedAddress]|None = None,
         spawn_method: str|None = None,
 
         # TODO: remove!
@@ -234,7 +227,7 @@ class Actor:
 
         # state
         self._cancel_complete = trio.Event()
-        self._cancel_called_by: tuple[str, tuple]|None = None
+        self._cancel_called_by_remote: tuple[str, tuple]|None = None
         self._cancel_called: bool = False
 
         # retreive and store parent `__main__` data which
@@ -256,12 +249,11 @@ class Actor:
         if arbiter_addr is not None:
             warnings.warn(
                 '`Actor(arbiter_addr=<blah>)` is now deprecated.\n'
-                'Use `registry_addrs: list[Address]` instead.',
+                'Use `registry_addrs: list[tuple]` instead.',
                 DeprecationWarning,
                 stacklevel=2,
             )
-
+            registry_addrs: list[UnwrappedAddress] = [arbiter_addr]
-            registry_addrs: list[Address] = [wrap_address(arbiter_addr)]
 
         # marked by the process spawning backend at startup
         # will be None for the parent most process started manually
@@ -300,10 +292,8 @@ class Actor:
         # input via the validator.
         self._reg_addrs: list[UnwrappedAddress] = []
         if registry_addrs:
-            _state._runtime_vars['_registry_addrs'] = self.reg_addrs = [
+            self.reg_addrs: list[UnwrappedAddress] = registry_addrs
-                addr.unwrap()
+            _state._runtime_vars['_registry_addrs'] = registry_addrs
-                for addr in registry_addrs
-            ]
 
     @property
     def aid(self) -> msgtypes.Aid:
@@ -349,125 +339,46 @@ class Actor:
     def pid(self) -> int:
         return self._aid.pid
 
-    @property
+    def pformat(self) -> str:
-    def repr_state(self) -> str:
+        ds: str = '='
-        if self.cancel_complete:
+        parent_uid: tuple|None = None
-            return 'cancelled'
-
-        elif canceller := self.cancel_caller:
-                return f' and cancel-called by {canceller}'
-
-        else:
-            return 'running'
-
-    def pformat(
-        self,
-        ds: str = ': ',
-        indent: int = 0,
-        privates: bool = False,
-    ) -> str:
-
-        fmtstr: str = f'|_id: {self.aid.reprol()!r}\n'
-        if privates:
-            aid_nest_prefix: str = '|_aid='
-            aid_field_repr: str = _pformat.nest_from_op(
-                input_op='',
-                text=pretty_struct.pformat(
-                    struct=self.aid,
-                    field_indent=2,
-                ),
-                op_suffix='',
-                nest_prefix=aid_nest_prefix,
-                nest_indent=0,
-            )
-            fmtstr: str = f'{aid_field_repr}'
-
         if rent_chan := self._parent_chan:
-            fmtstr += (
+            parent_uid = rent_chan.uid
-                f"|_parent{ds}{rent_chan.aid.reprol()}\n"
-            )
 
+        peers: list = []
         server: _server.IPCServer = self.ipc_server
         if server:
-            if privates:
+            peers: list[tuple] = list(server._peer_connected)
-                server_repr: str = self._ipc_server.pformat(
-                    privates=privates,
-                )
-                # create field ln as a key-header indented under
-                # and up to the section's key prefix.
-                # ^XXX if we were to indent `repr(Server)` to
-                # '<key>: '
-                #  _here_^
-                server_repr: str = _pformat.nest_from_op(
-                    input_op='',  # nest as sub-obj
-                    op_suffix='',
-                    text=server_repr,
-                )
-                fmtstr += (
-                    f"{server_repr}"
-                )
-            else:
-                fmtstr += (
-                    f'|_ipc: {server.repr_state!r}\n'
-                )
 
-        fmtstr += (
+        fmtstr: str = (
-            f'|_rpc: {len(self._rpc_tasks)} active tasks\n'
+            f' |_id: {self.aid!r}\n'
+            # f"   aid{ds}{self.aid!r}\n"
+            f"   parent{ds}{parent_uid}\n"
+            f'\n'
+            f' |_ipc: {len(peers)!r} connected peers\n'
+            f"   peers{ds}{peers!r}\n"
+            f"   ipc_server{ds}{self._ipc_server}\n"
+            f'\n'
+            f' |_rpc: {len(self._rpc_tasks)} tasks\n'
+            f"   ctxs{ds}{len(self._contexts)}\n"
+            f'\n'
+            f' |_runtime: ._task{ds}{self._task!r}\n'
+            f'   _spawn_method{ds}{self._spawn_method}\n'
+            f'   _actoruid2nursery{ds}{self._actoruid2nursery}\n'
+            f'   _forkserver_info{ds}{self._forkserver_info}\n'
+            f'\n'
+            f' |_state: "TODO: .repr_state()"\n'
+            f'   _cancel_complete{ds}{self._cancel_complete}\n'
+            f'   _cancel_called_by_remote{ds}{self._cancel_called_by_remote}\n'
+            f'   _cancel_called{ds}{self._cancel_called}\n'
         )
-
+        return (
-        # TODO, actually fix the .repr_state impl/output?
+            '<Actor(\n'
-        # append ipc-ctx state summary
+            +
-        # ctxs: dict = self._contexts
+            fmtstr
-        # if ctxs:
+            +
-        #     ctx_states: dict[str, int] = {}
+            ')>\n'
-        #     for ctx in self._contexts.values():
-        #         ctx_state: str = ctx.repr_state
-        #         cnt = ctx_states.setdefault(ctx_state, 0)
-        #         ctx_states[ctx_state] = cnt + 1
-
-        #     fmtstr += (
-        #         f"  ctxs{ds}{ctx_states}\n"
-        #     )
-
-        # runtime-state
-        task_name: str = '<dne>'
-        if task := self._task:
-            task_name: str = task.name
-        fmtstr += (
-            # TODO, this just like ctx?
-            f'|_state: {self.repr_state!r}\n'
-            f'  task: {task_name}\n'
-            f'  loglevel: {self.loglevel!r}\n'
-            f'  subactors_spawned: {len(self._actoruid2nursery)}\n'
         )
-        if not _state.is_root_process():
-            fmtstr += f'  spawn_method: {self._spawn_method!r}\n'
-
-        if privates:
-            fmtstr += (
-                # f'  actoruid2nursery{ds}{self._actoruid2nursery}\n'
-                f'  cancel_complete{ds}{self._cancel_complete}\n'
-                f'  cancel_called_by_remote{ds}{self._cancel_called_by}\n'
-                f'  cancel_called{ds}{self._cancel_called}\n'
-            )
-
-        if fmtstr:
-            fmtstr: str = textwrap.indent(
-                text=fmtstr,
-                prefix=' '*(1 + indent),
-            )
-
-        _repr: str = (
-            f'<{type(self).__name__}(\n'
-            f'{fmtstr}'
-            f')>\n'
-        )
-        if indent:
-            _repr: str = textwrap.indent(
-                text=_repr,
-                prefix=' '*indent,
-            )
-        return _repr
 
     __repr__ = pformat
 
@@ -475,11 +386,7 @@ class Actor:
     def reg_addrs(self) -> list[UnwrappedAddress]:
         '''
         List of (socket) addresses for all known (and contactable)
-        registry-service actors in "unwrapped" (i.e. IPC interchange
+        registry actors.
-        wire-compat) form.
-
-        If you are looking for the "wrapped" address form, use
-        `.registry_addrs` instead.
 
         '''
         return self._reg_addrs
@@ -498,14 +405,8 @@ class Actor:
 
         self._reg_addrs = addrs
 
-    @property
-    def registry_addrs(self) -> list[Address]:
-        return [wrap_address(uw_addr)
-                for uw_addr in self.reg_addrs]
-
     def load_modules(
         self,
-
     ) -> None:
         '''
         Load explicitly enabled python modules from local fs after
@@ -552,14 +453,6 @@ class Actor:
             )
             raise
 
-    # ?TODO, factor this meth-iface into a new `.rpc` subsys primitive?
-    # - _get_rpc_func(),
-    # - _deliver_ctx_payload(),
-    # - get_context(),
-    # - start_remote_task(),
-    # - cancel_rpc_tasks(),
-    # - _cancel_task(),
-    #
     def _get_rpc_func(self, ns, funcname):
         '''
         Try to lookup and return a target RPC func from the
@@ -603,11 +496,11 @@ class Actor:
         queue.
 
         '''
-        aid: msgtypes.Aid = chan.aid
+        uid: tuple[str, str] = chan.uid
-        assert aid, f"`chan.aid` can't be {aid}"
+        assert uid, f"`chan.uid` can't be {uid}"
         try:
             ctx: Context = self._contexts[(
-                aid.uid,
+                uid,
                 cid,
 
                 # TODO: how to determine this tho?
@@ -618,7 +511,7 @@ class Actor:
                 'Ignoring invalid IPC msg!?\n'
                 f'Ctx seems to not/no-longer exist??\n'
                 f'\n'
-                f'<=? {aid.reprol()!r}\n'
+                f'<=? {uid}\n'
                 f'  |_{pretty_struct.pformat(msg)}\n'
             )
             match msg:
@@ -667,7 +560,6 @@ class Actor:
           msging session's lifetime.
 
         '''
-        # ?TODO, use Aid here as well?
         actor_uid = chan.uid
         assert actor_uid
         try:
@@ -1016,22 +908,6 @@ class Actor:
             None,  # self cancel all rpc tasks
         )
 
-    @property
-    def cancel_complete(self) -> bool:
-        return self._cancel_complete.is_set()
-
-    @property
-    def cancel_called(self) -> bool:
-        '''
-        Was this actor requested to cancel by a remote peer actor.
-
-        '''
-        return self._cancel_called_by is not None
-
-    @property
-    def cancel_caller(self) -> msgtypes.Aid|None:
-        return self._cancel_called_by
-
     async def cancel(
         self,
 
@@ -1056,18 +932,20 @@ class Actor:
 
         '''
         (
-            requesting_aid,  # Aid
+            requesting_uid,
-            requester_type,  # str
+            requester_type,
             req_chan,
             log_meth,
         ) = (
-            req_chan.aid,
+            req_chan.uid,
             'peer',
             req_chan,
             log.cancel,
+
         ) if req_chan else (
+
             # a self cancel of ALL rpc tasks
-            self.aid,
+            self.uid,
             'self',
             self,
             log.runtime,
@@ -1075,14 +953,14 @@ class Actor:
         # TODO: just use the new `Context.repr_rpc: str` (and
         # other) repr fields instead of doing this all manual..
         msg: str = (
-            f'Actor-runtime cancel request from {requester_type!r}\n'
+            f'Actor-runtime cancel request from {requester_type}\n\n'
+            f'<=c) {requesting_uid}\n'
+            f'  |_{self}\n'
             f'\n'
-            f'<=c)\n'
-            f'{self}'
         )
 
         # TODO: what happens here when we self-cancel tho?
-        self._cancel_called_by: tuple = requesting_aid
+        self._cancel_called_by_remote: tuple = requesting_uid
         self._cancel_called = True
 
         # cancel all ongoing rpc tasks
@@ -1110,7 +988,7 @@ class Actor:
 
             # self-cancel **all** ongoing RPC tasks
             await self.cancel_rpc_tasks(
-                req_aid=requesting_aid,
+                req_uid=requesting_uid,
                 parent_chan=None,
             )
 
@@ -1127,11 +1005,19 @@ class Actor:
         self._cancel_complete.set()
         return True
 
+    # XXX: hard kill logic if needed?
+    # def _hard_mofo_kill(self):
+    #     # If we're the root actor or zombied kill everything
+    #     if self._parent_chan is None:  # TODO: more robust check
+    #         root = trio.lowlevel.current_root_task()
+    #         for n in root.child_nurseries:
+    #             n.cancel_scope.cancel()
+
     async def _cancel_task(
         self,
         cid: str,
         parent_chan: Channel,
-        requesting_aid: msgtypes.Aid|None,
+        requesting_uid: tuple[str, str]|None,
 
         ipc_msg: dict|None|bool = False,
 
@@ -1169,7 +1055,7 @@ class Actor:
             log.runtime(
                 'Cancel request for invalid RPC task.\n'
                 'The task likely already completed or was never started!\n\n'
-                f'<= canceller: {requesting_aid}\n'
+                f'<= canceller: {requesting_uid}\n'
                 f'=> {cid}@{parent_chan.uid}\n'
                 f'  |_{parent_chan}\n'
             )
@@ -1177,12 +1063,9 @@ class Actor:
 
         log.cancel(
             'Rxed cancel request for RPC task\n'
-            f'{ctx._task!r} <=c) {requesting_aid}\n'
+            f'<=c) {requesting_uid}\n'
-            f'|_>> {ctx.repr_rpc}\n'
+            f' |_{ctx._task}\n'
-
+            f'    >> {ctx.repr_rpc}\n'
-            # f'|_{ctx._task}\n'
-            # f'   >> {ctx.repr_rpc}\n'
-
             # f'=> {ctx._task}\n'
             # f'  >> Actor._cancel_task() => {ctx._task}\n'
             # f'  |_ {ctx._task}\n\n'
@@ -1203,9 +1086,9 @@ class Actor:
         )
         if (
             ctx._canceller is None
-            and requesting_aid
+            and requesting_uid
         ):
-            ctx._canceller: tuple = requesting_aid.uid
+            ctx._canceller: tuple = requesting_uid
 
         # TODO: pack the RPC `{'cmd': <blah>}` msg into a ctxc and
         # then raise and pack it here?
@@ -1231,7 +1114,7 @@ class Actor:
 
         # wait for _invoke to mark the task complete
         flow_info: str = (
-            f'<= canceller: {requesting_aid}\n'
+            f'<= canceller: {requesting_uid}\n'
             f'=> ipc-parent: {parent_chan}\n'
             f'|_{ctx}\n'
         )
@@ -1248,7 +1131,7 @@ class Actor:
 
     async def cancel_rpc_tasks(
         self,
-        req_aid: msgtypes.Aid,
+        req_uid: tuple[str, str],
 
         # NOTE: when None is passed we cancel **all** rpc
         # tasks running in this actor!
@@ -1265,7 +1148,7 @@ class Actor:
         if not tasks:
             log.runtime(
                 'Actor has no cancellable RPC tasks?\n'
-                f'<= canceller: {req_aid.reprol()}\n'
+                f'<= canceller: {req_uid}\n'
             )
             return
 
@@ -1305,7 +1188,7 @@ class Actor:
         )
         log.cancel(
             f'Cancelling {descr} RPC tasks\n\n'
-            f'<=c) {req_aid} [canceller]\n'
+            f'<=c) {req_uid} [canceller]\n'
             f'{rent_chan_repr}'
             f'c)=> {self.uid} [cancellee]\n'
             f'  |_{self} [with {len(tasks)} tasks]\n'
@@ -1333,7 +1216,7 @@ class Actor:
             await self._cancel_task(
                 cid,
                 task_caller_chan,
-                requesting_aid=req_aid,
+                requesting_uid=req_uid,
             )
 
         if tasks:
@@ -1361,13 +1244,25 @@ class Actor:
         '''
         return self.accept_addrs[0]
 
-    # TODO, this should delegate ONLY to the
+    def get_parent(self) -> Portal:
-    # `._spawn_spec._runtime_vars: dict` / `._state` APIs?
+        '''
-    #
+        Return a `Portal` to our parent.
-    # XXX, AH RIGHT that's why..
+
-    #   it's bc we pass this as a CLI flag to the child.py precisely
+        '''
-    #   bc we need the bootstrapping pre `async_main()`.. but maybe
+        assert self._parent_chan, "No parent channel for this actor?"
-    #   keep this as an impl deat and not part of the pub iface impl?
+        return Portal(self._parent_chan)
+
+    def get_chans(
+        self,
+        uid: tuple[str, str],
+
+    ) -> list[Channel]:
+        '''
+        Return all IPC channels to the actor with provided `uid`.
+
+        '''
+        return self._peers[uid]
+
     def is_infected_aio(self) -> bool:
         '''
         If `True`, this actor is running `trio` in guest mode on
@@ -1378,23 +1273,6 @@ class Actor:
         '''
         return self._infected_aio
 
-    # ?TODO, is this the right type for this method?
-    def get_parent(self) -> Portal:
-        '''
-        Return a `Portal` to our parent.
-
-        '''
-        assert self._parent_chan, "No parent channel for this actor?"
-        return Portal(self._parent_chan)
-
-    # XXX: hard kill logic if needed?
-    # def _hard_mofo_kill(self):
-    #     # If we're the root actor or zombied kill everything
-    #     if self._parent_chan is None:  # TODO: more robust check
-    #         root = trio.lowlevel.current_root_task()
-    #         for n in root.child_nurseries:
-    #             n.cancel_scope.cancel()
-
 
 async def async_main(
     actor: Actor,
@@ -1438,8 +1316,6 @@ async def async_main(
         # establish primary connection with immediate parent
         actor._parent_chan: Channel|None = None
 
-        # is this a sub-actor?
-        # get runtime info from parent.
         if parent_addr is not None:
             (
                 actor._parent_chan,
@@ -1474,18 +1350,18 @@ async def async_main(
         # parent is kept alive as a resilient service until
         # cancellation steps have (mostly) occurred in
         # a deterministic way.
-        root_tn: trio.Nursery
+        async with trio.open_nursery(
-        async with (
+            strict_exception_groups=False,
-            collapse_eg(),
+        ) as root_nursery:
-            trio.open_nursery() as root_tn,
+            actor._root_n = root_nursery
-        ):
-            actor._root_n = root_tn
             assert actor._root_n
 
             ipc_server: _server.IPCServer
             async with (
-                collapse_eg(),
+                trio.open_nursery(
-                trio.open_nursery() as service_nursery,
+                    strict_exception_groups=False,
+                ) as service_nursery,
+
                 _server.open_ipc_server(
                     parent_tn=service_nursery,
                     stream_handler_tn=service_nursery,
@@ -1536,6 +1412,9 @@ async def async_main(
 
                 # TODO: why is this not with the root nursery?
                 try:
+                    log.runtime(
+                        'Booting IPC server'
+                    )
                     eps: list = await ipc_server.listen_on(
                         accept_addrs=accept_addrs,
                         stream_handler_nursery=service_nursery,
@@ -1567,6 +1446,18 @@ async def async_main(
                 # TODO, just read direct from ipc_server?
                 accept_addrs: list[UnwrappedAddress] = actor.accept_addrs
 
+                # NOTE: only set the loopback addr for the 
+                # process-tree-global "root" mailbox since
+                # all sub-actors should be able to speak to
+                # their root actor over that channel.
+                if _state._runtime_vars['_is_root']:
+                    raddrs: list[Address] = _state._runtime_vars['_root_addrs']
+                    for addr in accept_addrs:
+                        waddr: Address = wrap_address(addr)
+                        raddrs.append(addr)
+                    else:
+                        _state._runtime_vars['_root_mailbox'] = raddrs[0]
+
                 # Register with the arbiter if we're told its addr
                 log.runtime(
                     f'Registering `{actor.name}` => {pformat(accept_addrs)}\n'
@@ -1584,7 +1475,6 @@ async def async_main(
                     except AssertionError:
                         await debug.pause()
 
-                    # !TODO, get rid of the local-portal crap XD
                     async with get_registry(addr) as reg_portal:
                         for accept_addr in accept_addrs:
                             accept_addr = wrap_address(accept_addr)
@@ -1609,7 +1499,7 @@ async def async_main(
                 # start processing parent requests until our channel
                 # server is 100% up and running.
                 if actor._parent_chan:
-                    await root_tn.start(
+                    await root_nursery.start(
                         partial(
                             _rpc.process_messages,
                             chan=actor._parent_chan,
@@ -1621,9 +1511,8 @@ async def async_main(
                     # 'Blocking on service nursery to exit..\n'
                 )
             log.runtime(
-                'Service nursery complete\n'
+                "Service nursery complete\n"
-                '\n'
+                "Waiting on root nursery to complete"
-                '->} waiting on root nursery to complete..\n'
             )
 
         # Blocks here as expected until the root nursery is
@@ -1678,7 +1567,6 @@ async def async_main(
     finally:
         teardown_report: str = (
             'Main actor-runtime task completed\n'
-            '\n'
         )
 
         # ?TODO? should this be in `._entry`/`._root` mods instead?
@@ -1720,8 +1608,7 @@ async def async_main(
         # Unregister actor from the registry-sys / registrar.
         if (
             is_registered
-            and
+            and not actor.is_registrar
-            not actor.is_registrar
         ):
             failed: bool = False
             for addr in actor.reg_addrs:
@@ -1756,8 +1643,7 @@ async def async_main(
             ipc_server.has_peers(check_chans=True)
         ):
             teardown_report += (
-                f'-> Waiting for remaining peers to clear..\n'
+                f'-> Waiting for remaining peers {ipc_server._peers} to clear..\n'
-                f'   {pformat(ipc_server._peers)}'
             )
             log.runtime(teardown_report)
             await ipc_server.wait_for_no_more_peers(
@@ -1765,23 +1651,15 @@ async def async_main(
             )
 
         teardown_report += (
-            '-]> all peer channels are complete.\n'
+            '-> All peer channels are complete\n'
         )
 
-    # op_nested_actor_repr: str = _pformat.nest_from_op(
-    #     input_op=')>',
-    #     text=actor.pformat(),
-    #     nest_prefix='|_',
-    #     nest_indent=1,  # under >
-    # )
     teardown_report += (
-        '-)> actor runtime main task exit.\n'
+        'Actor runtime exiting\n'
-        # f'{op_nested_actor_repr}'
+        f'>)\n'
+        f'|_{actor}\n'
     )
-    # if _state._runtime_vars['_is_root']:
+    log.info(teardown_report)
-    #     log.info(teardown_report)
-    # else:
-    log.runtime(teardown_report)
 
 
 # TODO: rename to `Registry` and move to `.discovery._registry`!

tractor/_spawn.py

@@ -34,9 +34,9 @@ from typing import (
 import trio
 from trio import TaskStatus
 
-from .devx import (
+from .devx.debug import (
-    debug,
+    maybe_wait_for_debugger,
-    pformat as _pformat
+    acquire_debug_lock,
 )
 from tractor._state import (
     current_actor,
@@ -51,17 +51,14 @@ from tractor._portal import Portal
 from tractor._runtime import Actor
 from tractor._entry import _mp_main
 from tractor._exceptions import ActorFailure
-from tractor.msg import (
+from tractor.msg.types import (
-    types as msgtypes,
+    Aid,
-    pretty_struct,
+    SpawnSpec,
 )
 
 
 if TYPE_CHECKING:
-    from ipc import (
+    from ipc import IPCServer
-        _server,
-        Channel,
-    )
     from ._supervise import ActorNursery
     ProcessType = TypeVar('ProcessType', mp.Process, trio.Process)
 
@@ -331,21 +328,20 @@ async def soft_kill(
     see `.hard_kill()`).
 
     '''
-    chan: Channel = portal.channel
+    peer_aid: Aid = portal.channel.aid
-    peer_aid: msgtypes.Aid = chan.aid
     try:
         log.cancel(
             f'Soft killing sub-actor via portal request\n'
             f'\n'
-            f'c)=> {peer_aid.reprol()}@[{chan.maddr}]\n'
+            f'(c=> {peer_aid}\n'
-            f'   |_{proc}\n'
+            f'  |_{proc}\n'
         )
         # wait on sub-proc to signal termination
         await wait_func(proc)
 
     except trio.Cancelled:
         with trio.CancelScope(shield=True):
-            await debug.maybe_wait_for_debugger(
+            await maybe_wait_for_debugger(
                 child_in_debug=_runtime_vars.get(
                     '_debug_mode', False
                 ),
@@ -469,7 +465,7 @@ async def trio_proc(
         "--uid",
         # TODO, how to pass this over "wire" encodings like
         # cmdline args?
-        # -[ ] maybe we can add an `msgtypes.Aid.min_tuple()` ?
+        # -[ ] maybe we can add an `Aid.min_tuple()` ?
         str(subactor.uid),
         # Address the child must connect to on startup
         "--parent_addr",
@@ -487,14 +483,13 @@ async def trio_proc(
 
     cancelled_during_spawn: bool = False
     proc: trio.Process|None = None
-    ipc_server: _server.Server = actor_nursery._actor.ipc_server
+    ipc_server: IPCServer = actor_nursery._actor.ipc_server
     try:
         try:
             proc: trio.Process = await trio.lowlevel.open_process(spawn_cmd, **proc_kwargs)
             log.runtime(
-                f'Started new child subproc\n'
+                'Started new child\n'
-                f'(>\n'
+                f'|_{proc}\n'
-                f' |_{proc}\n'
             )
 
             # wait for actor to spawn and connect back to us
@@ -512,10 +507,10 @@ async def trio_proc(
                 with trio.CancelScope(shield=True):
                     # don't clobber an ongoing pdb
                     if is_root_process():
-                        await debug.maybe_wait_for_debugger()
+                        await maybe_wait_for_debugger()
 
                     elif proc is not None:
-                        async with debug.acquire_debug_lock(subactor.uid):
+                        async with acquire_debug_lock(subactor.uid):
                             # soft wait on the proc to terminate
                             with trio.move_on_after(0.5):
                                 await proc.wait()
@@ -533,19 +528,14 @@ async def trio_proc(
 
         # send a "spawning specification" which configures the
         # initial runtime state of the child.
-        sspec = msgtypes.SpawnSpec(
+        sspec = SpawnSpec(
             _parent_main_data=subactor._parent_main_data,
             enable_modules=subactor.enable_modules,
             reg_addrs=subactor.reg_addrs,
             bind_addrs=bind_addrs,
             _runtime_vars=_runtime_vars,
         )
-        log.runtime(
+        log.runtime(f'Sending spawn spec: {str(sspec)}')
-            f'Sending spawn spec to child\n'
-            f'{{}}=> {chan.aid.reprol()!r}\n'
-            f'\n'
-            f'{pretty_struct.pformat(sspec)}\n'
-        )
         await chan.send(sspec)
 
         # track subactor in current nursery
@@ -573,7 +563,7 @@ async def trio_proc(
             # condition.
             await soft_kill(
                 proc,
-                trio.Process.wait,  # XXX, uses `pidfd_open()` below.
+                trio.Process.wait,
                 portal
             )
 
@@ -581,7 +571,8 @@ async def trio_proc(
             # tandem if not done already
             log.cancel(
                 'Cancelling portal result reaper task\n'
-                f'c)> {subactor.aid.reprol()!r}\n'
+                f'>c)\n'
+                f' |_{subactor.uid}\n'
             )
             nursery.cancel_scope.cancel()
 
@@ -590,24 +581,21 @@ async def trio_proc(
         # allowed! Do this **after** cancellation/teardown to avoid
         # killing the process too early.
         if proc:
-            reap_repr: str = _pformat.nest_from_op(
-                input_op='>x)',
-                text=subactor.pformat(),
-            )
             log.cancel(
                 f'Hard reap sequence starting for subactor\n'
-                f'{reap_repr}'
+                f'>x)\n'
+                f' |_{subactor}@{subactor.uid}\n'
             )
 
             with trio.CancelScope(shield=True):
                 # don't clobber an ongoing pdb
                 if cancelled_during_spawn:
                     # Try again to avoid TTY clobbering.
-                    async with debug.acquire_debug_lock(subactor.uid):
+                    async with acquire_debug_lock(subactor.uid):
                         with trio.move_on_after(0.5):
                             await proc.wait()
 
-                await debug.maybe_wait_for_debugger(
+                await maybe_wait_for_debugger(
                     child_in_debug=_runtime_vars.get(
                         '_debug_mode', False
                     ),
@@ -636,7 +624,7 @@ async def trio_proc(
                 #     acquire the lock and get notified of who has it,
                 #     check that uid against our known children?
                 # this_uid: tuple[str, str] = current_actor().uid
-                # await debug.acquire_debug_lock(this_uid)
+                # await acquire_debug_lock(this_uid)
 
                 if proc.poll() is None:
                     log.cancel(f"Attempting to hard kill {proc}")
@@ -739,7 +727,7 @@ async def mp_proc(
 
     log.runtime(f"Started {proc}")
 
-    ipc_server: _server.Server = actor_nursery._actor.ipc_server
+    ipc_server: IPCServer = actor_nursery._actor.ipc_server
     try:
         # wait for actor to spawn and connect back to us
         # channel should have handshake completed by the

tractor/_supervise.py

@@ -21,6 +21,7 @@
 from contextlib import asynccontextmanager as acm
 from functools import partial
 import inspect
+from pprint import pformat
 from typing import (
     TYPE_CHECKING,
 )
@@ -30,10 +31,7 @@ import warnings
 import trio
 
 
-from .devx import (
+from .devx.debug import maybe_wait_for_debugger
-    debug,
-    pformat as _pformat,
-)
 from ._addr import (
     UnwrappedAddress,
     mk_uuid,
@@ -42,11 +40,8 @@ from ._state import current_actor, is_main_process
 from .log import get_logger, get_loglevel
 from ._runtime import Actor
 from ._portal import Portal
-from .trionics import (
-    is_multi_cancelled,
-    collapse_eg,
-)
 from ._exceptions import (
+    is_multi_cancelled,
     ContextCancelled,
 )
 from ._root import (
@@ -202,7 +197,7 @@ class ActorNursery:
             loglevel=loglevel,
 
             # verbatim relay this actor's registrar addresses
-            registry_addrs=current_actor().registry_addrs,
+            registry_addrs=current_actor().reg_addrs,
         )
         parent_addr: UnwrappedAddress = self._actor.accept_addr
         assert parent_addr
@@ -327,10 +322,9 @@ class ActorNursery:
         server: IPCServer = self._actor.ipc_server
 
         with trio.move_on_after(3) as cs:
-            async with (
+            async with trio.open_nursery(
-                collapse_eg(),
+                strict_exception_groups=False,
-                trio.open_nursery() as tn,
+            ) as tn:
-            ):
 
                 subactor: Actor
                 proc: trio.Process
@@ -423,10 +417,10 @@ async def _open_and_supervise_one_cancels_all_nursery(
     # `ActorNursery.start_actor()`).
 
     # errors from this daemon actor nursery bubble up to caller
-    async with (
+    async with trio.open_nursery(
-        collapse_eg(),
+        strict_exception_groups=False,
-        trio.open_nursery() as da_nursery,
+        # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
-    ):
+    ) as da_nursery:
         try:
             # This is the inner level "run in actor" nursery. It is
             # awaited first since actors spawned in this way (using
@@ -436,10 +430,11 @@ async def _open_and_supervise_one_cancels_all_nursery(
             # immediately raised for handling by a supervisor strategy.
             # As such if the strategy propagates any error(s) upwards
             # the above "daemon actor" nursery will be notified.
-            async with (
+            async with trio.open_nursery(
-                collapse_eg(),
+                strict_exception_groups=False,
-                trio.open_nursery() as ria_nursery,
+                # ^XXX^ TODO? instead unpack any RAE as per "loose" style?
-            ):
+            ) as ria_nursery:
+
                 an = ActorNursery(
                     actor,
                     ria_nursery,
@@ -456,7 +451,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
                     # the "hard join phase".
                     log.runtime(
                         'Waiting on subactors to complete:\n'
-                        f'>}} {len(an._children)}\n'
+                        f'{pformat(an._children)}\n'
                     )
                     an._join_procs.set()
 
@@ -470,7 +465,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
                     # will make the pdb repl unusable.
                     # Instead try to wait for pdb to be released before
                     # tearing down.
-                    await debug.maybe_wait_for_debugger(
+                    await maybe_wait_for_debugger(
                         child_in_debug=an._at_least_one_child_in_debug
                     )
 
@@ -546,7 +541,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
 
             # XXX: yet another guard before allowing the cancel
             # sequence in case a (single) child is in debug.
-            await debug.maybe_wait_for_debugger(
+            await maybe_wait_for_debugger(
                 child_in_debug=an._at_least_one_child_in_debug
             )
 
@@ -595,14 +590,9 @@ async def _open_and_supervise_one_cancels_all_nursery(
     # final exit
 
 
-_shutdown_msg: str = (
-    'Actor-runtime-shutdown'
-)
-
-# @api_frame
 @acm
+# @api_frame
 async def open_nursery(
-    *,  # named params only!
     hide_tb: bool = True,
     **kwargs,
     # ^TODO, paramspec for `open_root_actor()`
@@ -687,26 +677,17 @@ async def open_nursery(
         ):
             __tracebackhide__: bool = False
 
-
+        msg: str = (
-        op_nested_an_repr: str = _pformat.nest_from_op(
+            'Actor-nursery exited\n'
-            input_op=')>',
+            f'|_{an}\n'
-            text=f'{an}',
-            # nest_prefix='|_',
-            nest_indent=1,  # under >
         )
-        an_msg: str = (
-            f'Actor-nursery exited\n'
-            f'{op_nested_an_repr}\n'
-        )
-        # keep noise low during std operation.
-        log.runtime(an_msg)
 
         if implicit_runtime:
             # shutdown runtime if it was started and report noisly
             # that we're did so.
-            msg: str = (
+            msg += '=> Shutting down actor runtime <=\n'
-                '\n'
-                '\n'
-                f'{_shutdown_msg} )>\n'
-            )
             log.info(msg)
+
+        else:
+            # keep noise low during std operation.
+            log.runtime(msg)

tractor/devx/debug/_post_mortem.py

@@ -59,7 +59,7 @@ from tractor._state import (
     debug_mode,
 )
 from tractor.log import get_logger
-from tractor.trionics import (
+from tractor._exceptions import (
     is_multi_cancelled,
 )
 from ._trace import (
@@ -148,59 +148,61 @@ def _post_mortem(
         repl_fixture=repl_fixture,
         boxed_maybe_exc=boxed_maybe_exc,
     )
-    if not enter_repl:
-        return
     try:
-        actor: Actor = current_actor()
+        if not enter_repl:
-        actor_repr: str = str(actor.uid)
+            # XXX, trigger `.release()` below immediately!
-        # ^TODO, instead a nice runtime-info + maddr + uid?
+            return
-        # -[ ] impl a `Actor.__repr()__`??
+        try:
-        #  |_ <task>:<thread> @ <actor>
+            actor: Actor = current_actor()
+            actor_repr: str = str(actor.uid)
+            # ^TODO, instead a nice runtime-info + maddr + uid?
+            # -[ ] impl a `Actor.__repr()__`??
+            #  |_ <task>:<thread> @ <actor>
 
-    except NoRuntime:
+        except NoRuntime:
-        actor_repr: str = '<no-actor-runtime?>'
+            actor_repr: str = '<no-actor-runtime?>'
 
-    try:
+        try:
-        task_repr: Task = trio.lowlevel.current_task()
+            task_repr: Task = trio.lowlevel.current_task()
-    except RuntimeError:
+        except RuntimeError:
-        task_repr: str = '<unknown-Task>'
+            task_repr: str = '<unknown-Task>'
 
-    # TODO: print the actor supervion tree up to the root
+        # TODO: print the actor supervion tree up to the root
-    # here! Bo
+        # here! Bo
-    log.pdb(
+        log.pdb(
-        f'{_crash_msg}\n'
+            f'{_crash_msg}\n'
-        f'x>(\n'
+            f'x>(\n'
-        f' |_ {task_repr} @ {actor_repr}\n'
+            f' |_ {task_repr} @ {actor_repr}\n'
 
-    )
+        )
 
-    # XXX NOTE(s) on `pdbp.xpm()` version..
+        # XXX NOTE(s) on `pdbp.xpm()` version..
-    #
+        #
-    # - seems to lose the up-stack tb-info?
+        # - seems to lose the up-stack tb-info?
-    # - currently we're (only) replacing this from `pdbp.xpm()`
+        # - currently we're (only) replacing this from `pdbp.xpm()`
-    #   to add the `end=''` to the print XD
+        #   to add the `end=''` to the print XD
-    #
+        #
-    print(traceback.format_exc(), end='')
+        print(traceback.format_exc(), end='')
-    caller_frame: FrameType = api_frame.f_back
+        caller_frame: FrameType = api_frame.f_back
 
-    # NOTE, see the impl details of these in the lib to
+        # NOTE, see the impl details of these in the lib to
-    # understand usage:
+        # understand usage:
-    # - `pdbp.post_mortem()`
+        # - `pdbp.post_mortem()`
-    # - `pdbp.xps()`
+        # - `pdbp.xps()`
-    # - `bdb.interaction()`
+        # - `bdb.interaction()`
-    repl.reset()
+        repl.reset()
-    repl.interaction(
+        repl.interaction(
-        frame=caller_frame,
+            frame=caller_frame,
-        # frame=None,
+            # frame=None,
-        traceback=tb,
+            traceback=tb,
-    )
+        )
-
+    finally:
-    # XXX NOTE XXX: this is abs required to avoid hangs!
+        # XXX NOTE XXX: this is abs required to avoid hangs!
-    #
+        #
-    # Since we presume the post-mortem was enaged to
+        # Since we presume the post-mortem was enaged to
-    # a task-ending error, we MUST release the local REPL request
+        # a task-ending error, we MUST release the local REPL request
-    # so that not other local task nor the root remains blocked!
+        # so that not other local task nor the root remains blocked!
-    DebugStatus.release()
+        DebugStatus.release()
 
 
 async def post_mortem(

tractor/devx/debug/_tty_lock.py

@@ -846,9 +846,9 @@ class DebugStatus:
 
         '''
         if not (
-            repl_fixture
-            or
             (rt_repl_fixture := _state._runtime_vars.get('repl_fixture'))
+            or
+            repl_fixture
         ):
             return True  # YES always enter
 

tractor/ipc/_chan.py

@@ -171,23 +171,11 @@ class Channel:
         )
         assert transport.raddr == addr
         chan = Channel(transport=transport)
-
+        log.runtime(
-        # ?TODO, compact this into adapter level-methods?
+            f'Connected channel IPC transport\n'
-        # -[ ] would avoid extra repr-calcs if level not active?
+            f'[>\n'
-        #   |_ how would the `calc_if_level` look though? func?
+            f' |_{chan}\n'
-        if log.at_least_level('runtime'):
+        )
-            from tractor.devx import (
-                pformat as _pformat,
-            )
-            chan_repr: str = _pformat.nest_from_op(
-                input_op='[>',
-                text=chan.pformat(),
-                nest_indent=1,
-            )
-            log.runtime(
-                f'Connected channel IPC transport\n'
-                f'{chan_repr}'
-            )
         return chan
 
     @cm
@@ -208,12 +196,9 @@ class Channel:
             self._transport.codec = orig
 
     # TODO: do a .src/.dst: str for maddrs?
-    def pformat(
+    def pformat(self) -> str:
-        self,
-        privates: bool = False,
-    ) -> str:
         if not self._transport:
-            return '<Channel( with inactive transport? )>'
+            return '<Channel with inactive transport?>'
 
         tpt: MsgTransport = self._transport
         tpt_name: str = type(tpt).__name__
@@ -221,35 +206,26 @@ class Channel:
             'connected' if self.connected()
             else 'closed'
         )
-        repr_str: str = (
+        return (
             f'<Channel(\n'
             f' |_status: {tpt_status!r}\n'
-        ) + (
             f'   _closed={self._closed}\n'
             f'   _cancel_called={self._cancel_called}\n'
-            if privates else ''
+            f'\n'
-        ) + (  # peer-actor (processs) section
+            f' |_peer: {self.aid}\n'
-            f' |_peer: {self.aid.reprol()!r}\n'
+            f'\n'
-            if self.aid else ' |_peer: <unknown>\n'
-        ) + (
             f' |_msgstream: {tpt_name}\n'
-            f'   maddr: {tpt.maddr!r}\n'
+            f'   proto={tpt.laddr.proto_key!r}\n'
-            f'   proto: {tpt.laddr.proto_key!r}\n'
+            f'   layer={tpt.layer_key!r}\n'
-            f'   layer: {tpt.layer_key!r}\n'
+            f'   laddr={tpt.laddr}\n'
-            f'   codec: {tpt.codec_key!r}\n'
+            f'   raddr={tpt.raddr}\n'
-            f'   .laddr={tpt.laddr}\n'
+            f'   codec={tpt.codec_key!r}\n'
-            f'   .raddr={tpt.raddr}\n'
+            f'   stream={tpt.stream}\n'
-        ) + (
+            f'   maddr={tpt.maddr!r}\n'
-            f'   ._transport.stream={tpt.stream}\n'
+            f'   drained={tpt.drained}\n'
-            f'   ._transport.drained={tpt.drained}\n'
-            if privates else ''
-        ) + (
             f'   _send_lock={tpt._send_lock.statistics()}\n'
-            if privates else ''
+            f')>\n'
-        ) + (
-            ')>\n'
         )
-        return repr_str
 
     # NOTE: making this return a value that can be passed to
     # `eval()` is entirely **optional** FYI!
@@ -271,10 +247,6 @@ class Channel:
     def raddr(self) -> Address|None:
         return self._transport.raddr if self._transport else None
 
-    @property
-    def maddr(self) -> str:
-        return self._transport.maddr if self._transport else '<no-tpt>'
-
     # TODO: something like,
     # `pdbp.hideframe_on(errors=[MsgTypeError])`
     # instead of the `try/except` hack we have rn..
@@ -462,8 +434,8 @@ class Channel:
         await self.send(aid)
         peer_aid: Aid = await self.recv()
         log.runtime(
-            f'Received hanshake with peer\n'
+            f'Received hanshake with peer actor,\n'
-            f'<= {peer_aid.reprol(sin_uuid=False)}\n'
+            f'{peer_aid}\n'
         )
         # NOTE, we always are referencing the remote peer!
         self.aid = peer_aid

tractor/ipc/_mp_bs.py

@@ -17,16 +17,9 @@
 Utils to tame mp non-SC madeness
 
 '''
-
-# !TODO! in 3.13 this can be disabled (the-same/similarly) using
-# a flag,
-# - [ ] soo if it works like this, drop this module entirely for
-#   3.13+ B)
-#  |_https://docs.python.org/3/library/multiprocessing.shared_memory.html
-#
 def disable_mantracker():
     '''
-    Disable all `multiprocessing` "resource tracking" machinery since
+    Disable all ``multiprocessing``` "resource tracking" machinery since
     it's an absolute multi-threaded mess of non-SC madness.
 
     '''

tractor/ipc/_server.py

@@ -26,7 +26,7 @@ from contextlib import (
 from functools import partial
 from itertools import chain
 import inspect
-import textwrap
+from pprint import pformat
 from types import (
     ModuleType,
 )
@@ -43,10 +43,7 @@ from trio import (
     SocketListener,
 )
 
-from ..devx.pformat import (
+# from ..devx import debug
-    ppfmt,
-    nest_from_op,
-)
 from .._exceptions import (
     TransportClosed,
 )
@@ -144,8 +141,9 @@ async def maybe_wait_on_canced_subs(
 
     ):
         log.cancel(
-            'Waiting on cancel request to peer\n'
+            'Waiting on cancel request to peer..\n'
-            f'c)=> {chan.aid.reprol()}@[{chan.maddr}]\n'
+            f'c)=>\n'
+            f'  |_{chan.aid}\n'
         )
 
         # XXX: this is a soft wait on the channel (and its
@@ -181,7 +179,7 @@ async def maybe_wait_on_canced_subs(
                 log.warning(
                     'Draining msg from disconnected peer\n'
                     f'{chan_info}'
-                    f'{ppfmt(msg)}\n'
+                    f'{pformat(msg)}\n'
                 )
                 # cid: str|None = msg.get('cid')
                 cid: str|None = msg.cid
@@ -250,7 +248,7 @@ async def maybe_wait_on_canced_subs(
                 if children := local_nursery._children:
                     # indent from above local-nurse repr
                     report += (
-                        f'   |_{ppfmt(children)}\n'
+                        f'   |_{pformat(children)}\n'
                     )
 
                 log.warning(report)
@@ -281,9 +279,8 @@ async def maybe_wait_on_canced_subs(
                     log.runtime(
                         f'Peer IPC broke but subproc is alive?\n\n'
 
-                        f'<=x {chan.aid.reprol()}@[{chan.maddr}]\n'
+                        f'<=x {chan.aid}@{chan.raddr}\n'
-                        f'\n'
+                        f'   |_{proc}\n'
-                        f'{proc}\n'
                     )
 
     return local_nursery
@@ -327,10 +324,9 @@ async def handle_stream_from_peer(
 
     chan = Channel.from_stream(stream)
     con_status: str = (
-        f'New inbound IPC transport connection\n'
+        'New inbound IPC connection <=\n'
-        f'<=( {stream!r}\n'
+        f'|_{chan}\n'
     )
-    con_status_steps: str = ''
 
     # initial handshake with peer phase
     try:
@@ -376,7 +372,7 @@ async def handle_stream_from_peer(
     if _pre_chan := server._peers.get(uid):
         familiar: str = 'pre-existing-peer'
     uid_short: str = f'{uid[0]}[{uid[1][-6:]}]'
-    con_status_steps += (
+    con_status += (
         f' -> Handshake with {familiar} `{uid_short}` complete\n'
     )
 
@@ -401,7 +397,7 @@ async def handle_stream_from_peer(
         None,
     )
     if event:
-        con_status_steps += (
+        con_status += (
             ' -> Waking subactor spawn waiters: '
             f'{event.statistics().tasks_waiting}\n'
             f' -> Registered IPC chan for child actor {uid}@{chan.raddr}\n'
@@ -412,7 +408,7 @@ async def handle_stream_from_peer(
         event.set()
 
     else:
-        con_status_steps += (
+        con_status += (
             f' -> Registered IPC chan for peer actor {uid}@{chan.raddr}\n'
         )  # type: ignore
 
@@ -426,15 +422,8 @@ async def handle_stream_from_peer(
     # TODO: can we just use list-ref directly?
     chans.append(chan)
 
-    con_status_steps += ' -> Entering RPC msg loop..\n'
+    con_status += ' -> Entering RPC msg loop..\n'
-    log.runtime(
+    log.runtime(con_status)
-        con_status
-        +
-        textwrap.indent(
-            con_status_steps,
-            prefix=' '*3,  # align to first-ln
-        )
-    )
 
     # Begin channel management - respond to remote requests and
     # process received reponses.
@@ -467,67 +456,41 @@ async def handle_stream_from_peer(
             disconnected=disconnected,
         )
 
-        # `Channel` teardown and closure sequence
+        # ``Channel`` teardown and closure sequence
         # drop ref to channel so it can be gc-ed and disconnected
-        #
-        # -[x]TODO mk this be like
-        # <=x Channel(
-        #     |_field: blah
-        # )>
-        op_repr: str = '<=x '
-        chan_repr: str = nest_from_op(
-            input_op=op_repr,
-            op_suffix='',
-            nest_prefix='',
-            text=chan.pformat(),
-            nest_indent=len(op_repr)-1,
-            rm_from_first_ln='<',
-        )
-
         con_teardown_status: str = (
-            f'IPC channel disconnect\n'
+            f'IPC channel disconnected:\n'
-            f'\n'
+            f'<=x uid: {chan.aid}\n'
-            f'{chan_repr}\n'
+            f'   |_{pformat(chan)}\n\n'
-            f'\n'
         )
-
         chans.remove(chan)
 
         # TODO: do we need to be this pedantic?
         if not chans:
             con_teardown_status += (
-                f'-> No more channels with {chan.aid.reprol()!r}\n'
+                f'-> No more channels with {chan.aid}'
             )
             server._peers.pop(uid, None)
 
-        if peers := list(server._peers.values()):
+        peers_str: str = ''
-            peer_cnt: int = len(peers)
+        for uid, chans in server._peers.items():
-            if (
+            peers_str += (
-                (first := peers[0][0]) is not chan
+                f'uid: {uid}\n'
-                and
+            )
-                not disconnected
+            for i, chan in enumerate(chans):
-                and
+                peers_str += (
-                peer_cnt > 1
+                    f' |_[{i}] {pformat(chan)}\n'
-            ):
-                con_teardown_status += (
-                    f'-> Remaining IPC {peer_cnt-1!r} peers:\n'
                 )
-                for chans in server._peers.values():
+
-                    first: Channel = chans[0]
+        con_teardown_status += (
-                    if not (
+            f'-> Remaining IPC {len(server._peers)} peers: {peers_str}\n'
-                        first is chan
+        )
-                        and
-                        disconnected
-                    ):
-                        con_teardown_status += (
-                            f'  |_{first.aid.reprol()!r} -> {len(chans)!r} chans\n'
-                        )
 
         # No more channels to other actors (at all) registered
         # as connected.
         if not server._peers:
             con_teardown_status += (
-                '-> Signalling no more peer connections!\n'
+                'Signalling no more peer channel connections'
             )
             server._no_more_peers.set()
 
@@ -616,10 +579,10 @@ async def handle_stream_from_peer(
 
 class Endpoint(Struct):
     '''
-    An instance of an IPC "bound" address where the lifetime of an
+    An instance of an IPC "bound" address where the lifetime of the
-    "ability to accept connections" and handle the subsequent
+    "ability to accept connections" (from clients) and then handle
-    sequence-of-packets (maybe oriented as sessions) is determined by
+    those inbound sessions or sequences-of-packets is determined by
-    the underlying nursery scope(s).
+    a (maybe pair of) nurser(y/ies).
 
     '''
     addr: Address
@@ -637,24 +600,6 @@ class Endpoint(Struct):
         MsgTransport,  # handle to encoded-msg transport stream
     ] = {}
 
-    def pformat(
-        self,
-        indent: int = 0,
-        privates: bool = False,
-    ) -> str:
-        type_repr: str = type(self).__name__
-        fmtstr: str = (
-            # !TODO, always be ns aware!
-            # f'|_netns: {netns}\n'
-            f' |.addr: {self.addr!r}\n'
-            f' |_peers: {len(self.peer_tpts)}\n'
-        )
-        return (
-            f'<{type_repr}(\n'
-            f'{fmtstr}'
-            f')>'
-        )
-
     async def start_listener(self) -> SocketListener:
         tpt_mod: ModuleType = inspect.getmodule(self.addr)
         lstnr: SocketListener = await tpt_mod.start_listener(
@@ -694,13 +639,11 @@ class Endpoint(Struct):
 class Server(Struct):
     _parent_tn: Nursery
     _stream_handler_tn: Nursery
-
     # level-triggered sig for whether "no peers are currently
     # connected"; field is **always** set to an instance but
     # initialized with `.is_set() == True`.
     _no_more_peers: trio.Event
 
-    # active eps as allocated by `.listen_on()`
     _endpoints: list[Endpoint] = []
 
     # connection tracking & mgmt
@@ -708,19 +651,12 @@ class Server(Struct):
         str,  # uaid
         list[Channel],  # IPC conns from peer
     ] = defaultdict(list)
-
-    # events-table with entries registered unset while the local
-    # actor is waiting on a new actor to inbound connect, often
-    # a parent waiting on its child just after spawn.
     _peer_connected: dict[
         tuple[str, str],
         trio.Event,
     ] = {}
 
     # syncs for setup/teardown sequences
-    # - null when not yet booted,
-    # - unset when active,
-    # - set when fully shutdown with 0 eps active.
     _shutdown: trio.Event|None = None
 
     # TODO, maybe just make `._endpoints: list[Endpoint]` and
@@ -728,6 +664,7 @@ class Server(Struct):
     # @property
     # def addrs2eps(self) -> dict[Address, Endpoint]:
     #     ...
+
     @property
     def proto_keys(self) -> list[str]:
         return [
@@ -753,7 +690,7 @@ class Server(Struct):
             # TODO: obvi a different server type when we eventually
             # support some others XD
             log.runtime(
-                f'Cancelling server(s) for tpt-protos\n'
+                f'Cancelling server(s) for\n'
                 f'{self.proto_keys!r}\n'
             )
             self._parent_tn.cancel_scope.cancel()
@@ -780,14 +717,6 @@ class Server(Struct):
                 f'protos: {tpt_protos!r}\n'
             )
 
-    def len_peers(
-        self,
-    ) -> int:
-        return len([
-            chan.connected()
-            for chan in chain(*self._peers.values())
-        ])
-
     def has_peers(
         self,
         check_chans: bool = False,
@@ -801,11 +730,13 @@ class Server(Struct):
             has_peers
             and
             check_chans
-            and
-            (peer_cnt := self.len_peers())
         ):
             has_peers: bool = (
-                peer_cnt > 0
+                any(chan.connected()
+                    for chan in chain(
+                        *self._peers.values()
+                    )
+                )
                 and
                 has_peers
             )
@@ -872,66 +803,30 @@ class Server(Struct):
 
         return ev.is_set()
 
-    @property
+    def pformat(self) -> str:
-    def repr_state(self) -> str:
-        '''
-        A `str`-status describing the current state of this
-        IPC server in terms of the current operating "phase".
-
-        '''
-        status = 'server is active'
-        if self.has_peers():
-            peer_cnt: int = self.len_peers()
-            status: str = (
-                f'{peer_cnt!r} peer chans'
-            )
-        else:
-            status: str = 'No peer chans'
-
-        if self.is_shutdown():
-            status: str = 'server-shutdown'
-
-        return status
-
-    def pformat(
-        self,
-        privates: bool = False,
-    ) -> str:
         eps: list[Endpoint] = self._endpoints
 
-        # state_repr: str = (
+        state_repr: str = (
-        #     f'{len(eps)!r} endpoints active'
+            f'{len(eps)!r} IPC-endpoints active'
-        # )
-        fmtstr = (
-            f' |_state: {self.repr_state!r}\n'
         )
-        if privates:
+        fmtstr = (
-            fmtstr += f'   no_more_peers: {self.has_peers()}\n'
+            f' |_state: {state_repr}\n'
-
+            f'   no_more_peers: {self.has_peers()}\n'
-            if self._shutdown is not None:
+        )
-                shutdown_stats: EventStatistics = self._shutdown.statistics()
+        if self._shutdown is not None:
-                fmtstr += (
+            shutdown_stats: EventStatistics = self._shutdown.statistics()
-                    f'   task_waiting_on_shutdown: {shutdown_stats}\n'
-                )
-
-        if eps := self._endpoints:
-            addrs: list[tuple] = [
-                ep.addr for ep in eps
-            ]
-            repr_eps: str = ppfmt(addrs)
-
             fmtstr += (
-                f' |_endpoints: {repr_eps}\n'
+                f'   task_waiting_on_shutdown: {shutdown_stats}\n'
-                # ^TODO? how to indent closing ']'..
             )
 
-        if peers := self._peers:
+        fmtstr += (
-            fmtstr += (
+            # TODO, use the `ppfmt()` helper from `modden`!
-                f' |_peers: {len(peers)} connected\n'
+            f' |_endpoints: {pformat(self._endpoints)}\n'
-            )
+            f' |_peers: {len(self._peers)} connected\n'
+        )
 
         return (
-            f'<Server(\n'
+            f'<IPCServer(\n'
             f'{fmtstr}'
             f')>\n'
         )
@@ -990,8 +885,8 @@ class Server(Struct):
             )
 
         log.runtime(
-            f'Binding endpoints\n'
+            f'Binding to endpoints for,\n'
-            f'{ppfmt(accept_addrs)}\n'
+            f'{accept_addrs}\n'
         )
         eps: list[Endpoint] = await self._parent_tn.start(
             partial(
@@ -1001,19 +896,13 @@ class Server(Struct):
                 listen_addrs=accept_addrs,
             )
         )
-        self._endpoints.extend(eps)
-
-        serv_repr: str = nest_from_op(
-            input_op='(>',
-            text=self.pformat(),
-            nest_indent=1,
-        )
         log.runtime(
-            f'Started IPC server\n'
+            f'Started IPC endpoints\n'
-            f'{serv_repr}'
+            f'{eps}\n'
         )
 
-        # XXX, a little sanity on new ep allocations
+        self._endpoints.extend(eps)
+        # XXX, just a little bit of sanity
         group_tn: Nursery|None = None
         ep: Endpoint
         for ep in eps:
@@ -1067,13 +956,9 @@ async def _serve_ipc_eps(
                     stream_handler_tn=stream_handler_tn,
                 )
                 try:
-                    ep_sclang: str = nest_from_op(
-                        input_op='>[',
-                        text=f'{ep.pformat()}',
-                    )
                     log.runtime(
                         f'Starting new endpoint listener\n'
-                        f'{ep_sclang}\n'
+                        f'{ep}\n'
                     )
                     listener: trio.abc.Listener = await ep.start_listener()
                     assert listener is ep._listener
@@ -1111,6 +996,17 @@ async def _serve_ipc_eps(
                     handler_nursery=stream_handler_tn
                 )
             )
+            # TODO, wow make this message better! XD
+            log.runtime(
+                'Started server(s)\n'
+                +
+                '\n'.join([f'|_{addr}' for addr in listen_addrs])
+            )
+
+            log.runtime(
+                f'Started IPC endpoints\n'
+                f'{eps}\n'
+            )
             task_status.started(
                 eps,
             )
@@ -1153,7 +1049,8 @@ async def open_ipc_server(
         try:
             yield ipc_server
             log.runtime(
-                'Server-tn running until terminated\n'
+                f'Waiting on server to shutdown or be cancelled..\n'
+                f'{ipc_server}'
             )
             # TODO? when if ever would we want/need this?
             # with trio.CancelScope(shield=True):

tractor/ipc/_shm.py

@@ -789,11 +789,6 @@ def open_shm_list(
         readonly=readonly,
     )
 
-    # TODO, factor into a @actor_fixture acm-API?
-    # -[ ] also `@maybe_actor_fixture()` which inludes
-    #     the .current_actor() convenience check?
-    #   |_ orr can that just be in the sin-maybe-version?
-    #
     # "close" attached shm on actor teardown
     try:
         actor = tractor.current_actor()

tractor/ipc/_tcp.py

@@ -160,9 +160,10 @@ async def start_listener(
     Start a TCP socket listener on the given `TCPAddress`.
 
     '''
-    log.runtime(
+    log.info(
-        f'Trying socket bind\n'
+        f'Attempting to bind TCP socket\n'
-        f'>[ {addr}\n'
+        f'>[\n'
+        f'|_{addr}\n'
     )
     # ?TODO, maybe we should just change the lower-level call this is
     # using internall per-listener?
@@ -177,10 +178,11 @@ async def start_listener(
     assert len(listeners) == 1
     listener = listeners[0]
     host, port = listener.socket.getsockname()[:2]
-    bound_addr: TCPAddress = type(addr).from_addr((host, port))
+
     log.info(
         f'Listening on TCP socket\n'
-        f'[> {bound_addr}\n'
+        f'[>\n'
+        f' |_{addr}\n'
     )
     return listener
 

tractor/log.py

@@ -81,35 +81,10 @@ BOLD_PALETTE = {
 }
 
 
-def at_least_level(
-    log: Logger|LoggerAdapter,
-    level: int|str,
-) -> bool:
-    '''
-    Predicate to test if a given level is active.
-
-    '''
-    if isinstance(level, str):
-        level: int = CUSTOM_LEVELS[level.upper()]
-
-    if log.getEffectiveLevel() <= level:
-        return True
-    return False
-
-
 # TODO: this isn't showing the correct '{filename}'
 # as it did before..
 class StackLevelAdapter(LoggerAdapter):
 
-    def at_least_level(
-        self,
-        level: str,
-    ) -> bool:
-        return at_least_level(
-            log=self,
-            level=level,
-        )
-
     def transport(
         self,
         msg: str,
@@ -426,3 +401,19 @@ def get_loglevel() -> str:
 
 # global module logger for tractor itself
 log: StackLevelAdapter = get_logger('tractor')
+
+
+def at_least_level(
+    log: Logger|LoggerAdapter,
+    level: int|str,
+) -> bool:
+    '''
+    Predicate to test if a given level is active.
+
+    '''
+    if isinstance(level, str):
+        level: int = CUSTOM_LEVELS[level.upper()]
+
+    if log.getEffectiveLevel() <= level:
+        return True
+    return False

tractor/msg/_ops.py

@@ -210,14 +210,12 @@ class PldRx(Struct):
         match msg:
             case Return()|Error():
                 log.runtime(
-                    f'Rxed final-outcome msg\n'
+                    f'Rxed final outcome msg\n'
-                    f'\n'
                     f'{msg}\n'
                 )
             case Stop():
                 log.runtime(
                     f'Rxed stream stopped msg\n'
-                    f'\n'
                     f'{msg}\n'
                 )
                 if passthrough_non_pld_msgs:
@@ -263,9 +261,8 @@ class PldRx(Struct):
         if (
             type(msg) is Return
         ):
-            log.runtime(
+            log.info(
                 f'Rxed final result msg\n'
-                f'\n'
                 f'{msg}\n'
             )
         return self.decode_pld(
@@ -307,13 +304,10 @@ class PldRx(Struct):
                 try:
                     pld: PayloadT = self._pld_dec.decode(pld)
                     log.runtime(
-                        f'Decoded payload for\n'
+                        'Decoded msg payload\n\n'
-                        # f'\n'
                         f'{msg}\n'
-                        # ^TODO?, ideally just render with `,
+                        f'where payload decoded as\n'
-                        # pld={decode}` in the `msg.pformat()`??
+                        f'|_pld={pld!r}\n'
-                        f'where, '
-                        f'{type(msg).__name__}.pld={pld!r}\n'
                     )
                     return pld
                 except TypeError as typerr:
@@ -500,8 +494,7 @@ def limit_plds(
 
     finally:
         log.runtime(
-            f'Reverted to previous payload-decoder\n'
+            'Reverted to previous payload-decoder\n\n'
-            f'\n'
             f'{orig_pldec}\n'
         )
         # sanity on orig settings
@@ -636,8 +629,7 @@ async def drain_to_final_msg(
                     (local_cs := rent_n.cancel_scope).cancel_called
                 ):
                     log.cancel(
-                        f'RPC-ctx cancelled by local-parent scope during drain!\n'
+                        'RPC-ctx cancelled by local-parent scope during drain!\n\n'
-                        f'\n'
                         f'c}}>\n'
                         f' |_{rent_n}\n'
                         f'   |_.cancel_scope = {local_cs}\n'
@@ -671,8 +663,7 @@ async def drain_to_final_msg(
             # final result arrived!
             case Return():
                 log.runtime(
-                    f'Context delivered final draining msg\n'
+                    'Context delivered final draining msg:\n'
-                    f'\n'
                     f'{pretty_struct.pformat(msg)}'
                 )
                 ctx._result: Any = pld
@@ -706,14 +697,12 @@ async def drain_to_final_msg(
                 ):
                     log.cancel(
                         'Cancelling `MsgStream` drain since '
-                        f'{reason}\n'
+                        f'{reason}\n\n'
-                        f'\n'
                         f'<= {ctx.chan.uid}\n'
-                        f'  |_{ctx._nsf}()\n'
+                        f'  |_{ctx._nsf}()\n\n'
-                        f'\n'
                         f'=> {ctx._task}\n'
-                        f'  |_{ctx._stream}\n'
+                        f'  |_{ctx._stream}\n\n'
-                        f'\n'
+
                         f'{pretty_struct.pformat(msg)}\n'
                     )
                     break
@@ -750,8 +739,7 @@ async def drain_to_final_msg(
             case Stop():
                 pre_result_drained.append(msg)
                 log.runtime(  # normal/expected shutdown transaction
-                    f'Remote stream terminated due to "stop" msg\n'
+                    'Remote stream terminated due to "stop" msg:\n\n'
-                    f'\n'
                     f'{pretty_struct.pformat(msg)}\n'
                 )
                 continue
@@ -826,8 +814,7 @@ async def drain_to_final_msg(
 
     else:
         log.cancel(
-            f'Skipping `MsgStream` drain since final outcome is set\n'
+            'Skipping `MsgStream` drain since final outcome is set\n\n'
-            f'\n'
             f'{ctx.outcome}\n'
         )
 

tractor/msg/types.py

@@ -154,39 +154,6 @@ class Aid(
     #     should also include at least `.pid` (equiv to port for tcp)
     #     and/or host-part always?
 
-    @property
-    def uid(self) -> tuple[str, str]:
-        '''
-        Legacy actor "unique-id" pair format.
-
-        '''
-        return (
-            self.name,
-            self.uuid,
-        )
-
-    def reprol(
-        self,
-        sin_uuid: bool = True,
-    ) -> str:
-        if not sin_uuid:
-            return (
-                f'{self.name}[{self.uuid[:6]}]@{self.pid!r}'
-            )
-        return (
-            f'{self.name}@{self.pid!r}'
-        )
-
-    # mk hashable via `.uuid`
-    def __hash__(self) -> int:
-        return hash(self.uuid)
-
-    def __eq__(self, other: Aid) -> bool:
-        return self.uuid == other.uuid
-
-    # use pretty fmt since often repr-ed for console/log
-    __repr__ = pretty_struct.Struct.__repr__
-
 
 class SpawnSpec(
     pretty_struct.Struct,

tractor/to_asyncio.py

@@ -38,6 +38,7 @@ from typing import (
 import tractor
 from tractor._exceptions import (
     InternalError,
+    is_multi_cancelled,
     TrioTaskExited,
     TrioCancelled,
     AsyncioTaskExited,
@@ -58,9 +59,6 @@ from tractor.log import (
 # from tractor.msg import (
 #     pretty_struct,
 # )
-from tractor.trionics import (
-    is_multi_cancelled,
-)
 from tractor.trionics._broadcast import (
     broadcast_receiver,
     BroadcastReceiver,

tractor/trionics/__init__.py

@@ -32,8 +32,4 @@ from ._broadcast import (
 from ._beg import (
     collapse_eg as collapse_eg,
     maybe_collapse_eg as maybe_collapse_eg,
-    is_multi_cancelled as is_multi_cancelled,
-)
-from ._taskc import (
-    maybe_raise_from_masking_exc as maybe_raise_from_masking_exc,
 )

tractor/trionics/_beg.py

@@ -22,16 +22,11 @@ first-class-`trio` from a historical perspective B)
 from contextlib import (
     asynccontextmanager as acm,
 )
-from typing import (
-    Literal,
-)
-
-import trio
 
 
 def maybe_collapse_eg(
     beg: BaseExceptionGroup,
-) -> BaseException|bool:
+) -> BaseException:
     '''
     If the input beg can collapse to a single non-eg sub-exception,
     return it instead.
@@ -40,13 +35,11 @@ def maybe_collapse_eg(
     if len(excs := beg.exceptions) == 1:
         return excs[0]
 
-    return False
+    return beg
 
 
 @acm
-async def collapse_eg(
+async def collapse_eg():
-    hide_tb: bool = True,
-):
     '''
     If `BaseExceptionGroup` raised in the body scope is
     "collapse-able" (in the same way that
@@ -54,75 +47,12 @@ async def collapse_eg(
     only raise the lone emedded non-eg in in place.
 
     '''
-    __tracebackhide__: bool = hide_tb
     try:
         yield
     except* BaseException as beg:
         if (
             exc := maybe_collapse_eg(beg)
-        ):
+        ) is not beg:
-            if cause := exc.__cause__:
-                raise exc from cause
-
             raise exc
 
         raise beg
-
-
-def is_multi_cancelled(
-    beg: BaseException|BaseExceptionGroup,
-
-    ignore_nested: set[BaseException] = set(),
-
-) -> Literal[False]|BaseExceptionGroup:
-    '''
-    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 not in ignore_nested
-        # XXX always count-in `trio`'s native signal
-    ):
-        ignore_nested.update({trio.Cancelled})
-
-    if isinstance(beg, BaseExceptionGroup):
-        # https://docs.python.org/3/library/exceptions.html#BaseExceptionGroup.subgroup
-        # |_ "The condition can be an exception type or tuple of
-        #   exception types, in which case each exception is checked
-        #   for a match using the same check that is used in an
-        #   except clause. The condition can also be a callable
-        #   (other than a type object) that accepts an exception as
-        #   its single argument and returns true for the exceptions
-        #   that should be in the subgroup."
-        matched_exc: BaseExceptionGroup|None = beg.subgroup(
-            tuple(ignore_nested),
-
-            # ??TODO, complain about why not allowed to use
-            # named arg style calling???
-            # XD .. wtf?
-            # 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/trionics/_mngrs.py

@@ -40,8 +40,6 @@ from typing import (
 import trio
 from tractor._state import current_actor
 from tractor.log import get_logger
-from ._beg import collapse_eg
-
 
 if TYPE_CHECKING:
     from tractor import ActorNursery
@@ -72,7 +70,8 @@ async def maybe_open_nursery(
         yield nursery
     else:
         async with lib.open_nursery(**kwargs) as nursery:
-            nursery.cancel_scope.shield = shield
+            if lib == trio:
+                nursery.cancel_scope.shield = shield
             yield nursery
 
 
@@ -113,19 +112,17 @@ async def gather_contexts(
     None,
 ]:
     '''
-    Concurrently enter a sequence of async context managers (`acm`s),
+    Concurrently enter a sequence of async context managers (acms),
-    each scheduled in a separate `trio.Task` and deliver their
+    each from a separate `trio` task and deliver the unwrapped
-    unwrapped `yield`-ed values in the same order once all `@acm`s
+    `yield`-ed values in the same order once all managers have entered.
-    in every task have entered.
 
-    On exit, all `acm`s are subsequently and concurrently exited with
+    On exit, all acms are subsequently and concurrently exited.
-    **no order guarantees**.
 
     This function is somewhat similar to a batch of non-blocking
     calls to `contextlib.AsyncExitStack.enter_async_context()`
     (inside a loop) *in combo with* a `asyncio.gather()` to get the
     `.__aenter__()`-ed values, except the managers are both
-    concurrently entered and exited and *cancellation-just-works™*.
+    concurrently entered and exited and *cancellation just works*(R).
 
     '''
     seed: int = id(mngrs)
@@ -145,15 +142,16 @@ async def gather_contexts(
     if not mngrs:
         raise ValueError(
             '`.trionics.gather_contexts()` input mngrs is empty?\n'
-            '\n'
             'Did try to use inline generator syntax?\n'
-            'Use a non-lazy iterator or sequence-type intead!\n'
+            'Use a non-lazy iterator or sequence type intead!'
         )
 
-    async with (
+    async with trio.open_nursery(
-        collapse_eg(),
+        strict_exception_groups=False,
-        trio.open_nursery() as tn,
+        # ^XXX^ TODO? soo roll our own then ??
-    ):
+        # -> since we kinda want the "if only one `.exception` then
+        # just raise that" interface?
+    ) as tn:
         for mngr in mngrs:
             tn.start_soon(
                 _enter_and_wait,
@@ -170,7 +168,7 @@ async def gather_contexts(
         try:
             yield tuple(unwrapped.values())
         finally:
-            # XXX NOTE: this is ABSOLUTELY REQUIRED to avoid
+            # NOTE: this is ABSOLUTELY REQUIRED to avoid
             # the following wacky bug:
             # <tractorbugurlhere>
             parent_exit.set()

tractor/trionics/_taskc.py

@@ -1,185 +0,0 @@
-# tractor: structured concurrent "actors".
-# Copyright 2018-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/>.
-
-'''
-`trio.Task` cancellation helpers, extensions and "holsters".
-
-'''
-from __future__ import annotations
-from contextlib import (
-    asynccontextmanager as acm,
-)
-from typing import TYPE_CHECKING
-
-import trio
-from tractor.log import get_logger
-
-log = get_logger(__name__)
-
-
-if TYPE_CHECKING:
-    from tractor.devx.debug import BoxedMaybeException
-
-
-def find_masked_excs(
-    maybe_masker: BaseException,
-    unmask_from: set[BaseException],
-) -> BaseException|None:
-    ''''
-    Deliver any `maybe_masker.__context__` provided
-    it a declared masking exc-type entry in `unmask_from`.
-
-    '''
-    if (
-        type(maybe_masker) in unmask_from
-        and
-        (exc_ctx := maybe_masker.__context__)
-
-        # TODO? what about any cases where
-        # they could be the same type but not same instance?
-        # |_i.e. a cancel masking a cancel ??
-        # or (
-        #     exc_ctx is not maybe_masker
-        # )
-    ):
-        return exc_ctx
-
-    return None
-
-
-# XXX, relevant ish discussion @ `trio`-core,
-# https://github.com/python-trio/trio/issues/455#issuecomment-2785122216
-#
-@acm
-async def maybe_raise_from_masking_exc(
-    tn: trio.Nursery|None = None,
-    unmask_from: (
-        BaseException|
-        tuple[BaseException]
-    ) = (trio.Cancelled,),
-
-    raise_unmasked: bool = True,
-    extra_note: str = (
-        'This can occurr when,\n'
-        ' - a `trio.Nursery` scope embeds a `finally:`-block '
-        'which executes a checkpoint!'
-        #
-        # ^TODO? other cases?
-    ),
-
-    always_warn_on: tuple[BaseException] = (
-        trio.Cancelled,
-    ),
-    # ^XXX, special case(s) where we warn-log bc likely
-    # there will be no operational diff since the exc
-    # is always expected to be consumed.
-) -> BoxedMaybeException:
-    '''
-    Maybe un-mask and re-raise exception(s) suppressed by a known
-    error-used-as-signal type (cough namely `trio.Cancelled`).
-
-    Though this unmasker targets cancelleds, it can be used more
-    generally to capture and unwrap masked excs detected as
-    `.__context__` values which were suppressed by any error type
-    passed in `unmask_from`.
-
-    -------------
-    STILL-TODO ??
-    -------------
-    -[ ] support for egs which have multiple masked entries in
-        `maybe_eg.exceptions`, in which case we should unmask the
-        individual sub-excs but maintain the eg-parent's form right?
-
-    '''
-    from tractor.devx.debug import (
-        BoxedMaybeException,
-        pause,
-    )
-    boxed_maybe_exc = BoxedMaybeException(
-        raise_on_exit=raise_unmasked,
-    )
-    matching: list[BaseException]|None = None
-    maybe_eg: ExceptionGroup|None
-    maybe_eg: ExceptionGroup|None
-
-    if tn:
-        try:  # handle egs
-            yield boxed_maybe_exc
-            return
-        except* unmask_from as _maybe_eg:
-            maybe_eg = _maybe_eg
-            matches: ExceptionGroup
-            matches, _ = maybe_eg.split(
-                unmask_from
-            )
-            if not matches:
-                raise
-
-            matching: list[BaseException] = matches.exceptions
-    else:
-        try:  # handle non-egs
-            yield boxed_maybe_exc
-            return
-        except unmask_from as _maybe_exc:
-            maybe_exc = _maybe_exc
-            matching: list[BaseException] = [
-                maybe_exc
-            ]
-
-        # XXX, only unmask-ed for debuggin!
-        # TODO, remove eventually..
-        except BaseException as _berr:
-            berr = _berr
-            await pause(shield=True)
-            raise berr
-
-    if matching is None:
-        raise
-
-    masked: list[tuple[BaseException, BaseException]] = []
-    for exc_match in matching:
-
-        if exc_ctx := find_masked_excs(
-            maybe_masker=exc_match,
-            unmask_from={unmask_from},
-        ):
-            masked.append((exc_ctx, exc_match))
-            boxed_maybe_exc.value = exc_match
-            note: str = (
-                f'\n'
-                f'^^WARNING^^ the above {exc_ctx!r} was masked by a {unmask_from!r}\n'
-            )
-            if extra_note:
-                note += (
-                    f'\n'
-                    f'{extra_note}\n'
-                )
-            exc_ctx.add_note(note)
-
-            if type(exc_match) in always_warn_on:
-                log.warning(note)
-
-            # await tractor.pause(shield=True)
-            if raise_unmasked:
-
-                if len(masked) < 2:
-                    raise exc_ctx from exc_match
-                else:
-                    # ?TODO, see above but, possibly unmasking sub-exc
-                    # entries if there are > 1
-                    await pause(shield=True)
-    else:
-        raise

uv.lock

@@ -417,7 +417,7 @@ dev = [
     { name = "pyperclip", specifier = ">=1.9.0" },
     { name = "pytest", specifier = ">=8.3.5" },
     { name = "stackscope", specifier = ">=0.2.2,<0.3" },
-    { name = "typing-extensions", specifier = ">=4.13.2" },
+    { name = "typing-extensions", specifier = ">=4.14.1" },
     { name = "xonsh", specifier = ">=0.19.2" },
 ]
 
@@ -452,11 +452,11 @@ wheels = [
 
 [[package]]
 name = "typing-extensions"
-version = "4.13.2"
+version = "4.14.1"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/f6/37/23083fcd6e35492953e8d2aaaa68b860eb422b34627b13f2ce3eb6106061/typing_extensions-4.13.2.tar.gz", hash = "sha256:e6c81219bd689f51865d9e372991c540bda33a0379d5573cddb9a3a23f7caaef", size = 106967, upload-time = "2025-04-10T14:19:05.416Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/98/5a/da40306b885cc8c09109dc2e1abd358d5684b1425678151cdaed4731c822/typing_extensions-4.14.1.tar.gz", hash = "sha256:38b39f4aeeab64884ce9f74c94263ef78f3c22467c8724005483154c26648d36", size = 107673, upload-time = "2025-07-04T13:28:34.16Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/8b/54/b1ae86c0973cc6f0210b53d508ca3641fb6d0c56823f288d108bc7ab3cc8/typing_extensions-4.13.2-py3-none-any.whl", hash = "sha256:a439e7c04b49fec3e5d3e2beaa21755cadbbdc391694e28ccdd36ca4a1408f8c", size = 45806, upload-time = "2025-04-10T14:19:03.967Z" },
+    { url = "https://files.pythonhosted.org/packages/b5/00/d631e67a838026495268c2f6884f3711a15a9a2a96cd244fdaea53b823fb/typing_extensions-4.14.1-py3-none-any.whl", hash = "sha256:d1e1e3b58374dc93031d6eda2420a48ea44a36c2b4766a4fdeb3710755731d76", size = 43906, upload-time = "2025-07-04T13:28:32.743Z" },
 ]
 
 [[package]]