Mega-refactor on `._invoke()` targeting `@context`s

Since eventually we want to implement all other RPC "func types" as
contexts underneath this starts the rework to move all the other cases
into a separate func not only to simplify the main `._invoke()` body but
also as a reminder of the intention to do it XD

Details of re-factor:
- add a new `._invoke_non_context()` which just moves all the old blocks
  for non-context handling to a single def.
- factor what was basically just the `finally:` block handler (doing all
  the task bookkeeping) into a new `@acm`: `_errors_relayed_via_ipc()`
  with that content packed into the post-`yield` (also with a `hide_tb:
  bool` flag added of course).
  * include a `debug_kbis: bool` for when needed.
- since the `@context` block is the only type left in the main
  `_invoke()` body, de-dent it so it's more grok-able B)

Obviously this patch also includes a few improvements regarding
context-cancellation-semantics (for the `context` RPC case) on the
callee side in order to match previous changes to the `Context` api:
- always setting any ctxc as the `Context._local_error`.
- using the new convenience `.maybe_raise()` topically (for now).
- avoiding any previous reliance on `Context.cancelled_caught` for
  anything public of meaning.

Further included is more logging content updates:
- being pedantic in `.cancel()` msgs about whether termination is caused
  by error or ctxc.
- optional `._invoke()` traceback hiding via a `hide_tb: bool`.
- simpler log headers throughout instead leveraging new `.__repr__()` on
  primitives.
- buncha `<= <actor-uid>` sent some message emissions.
- simplified handshake statuses reporting.

Other subsys api changes we need to match:
- change to `Channel.transport`.
- avoiding any `local_nursery: ActorNursery` waiting when the
  `._implicit_runtime_started` is set.

And yes, lotsa more comments for #TODOs dawg.. since there's always
somethin!

tests/test_advanced_streaming.py

@@ -329,7 +329,7 @@ async def inf_streamer(
 
             # close out the stream gracefully
             except trio.ClosedResourceError:
-                print('transport closed on streamer side!')
+                print('msgstream closed on streamer side!')
                 assert stream.closed
                 break
         else:

tests/test_child_manages_service_nursery.py

@@ -10,6 +10,7 @@ from contextlib import asynccontextmanager as acm
 
 import pytest
 import trio
+from trio_typing import TaskStatus
 import tractor
 from tractor import RemoteActorError
 from async_generator import aclosing

tests/test_debugger.py

@@ -10,13 +10,12 @@ TODO:
     - wonder if any of it'll work on OS X?
 
 """
-from functools import partial
 import itertools
-# from os import path
+from os import path
 from typing import Optional
 import platform
 import pathlib
-# import sys
+import sys
 import time
 
 import pytest
@@ -26,10 +25,6 @@ from pexpect.exceptions import (
     EOF,
 )
 
-from tractor.devx._debug import (
-    _pause_msg,
-    _crash_msg,
-)
 from conftest import (
     examples_dir,
     _ci_env,
@@ -128,52 +123,20 @@ def expect(
         raise
 
 
-def in_prompt_msg(
-    prompt: str,
-    parts: list[str],
-
-    pause_on_false: bool = False,
-    print_prompt_on_false: bool = True,
-
-) -> bool:
-    '''
-    Predicate check if (the prompt's) std-streams output has all
-    `str`-parts in it.
-
-    Can be used in test asserts for bulk matching expected
-    log/REPL output for a given `pdb` interact point.
-
-    '''
-    for part in parts:
-        if part not in prompt:
-
-            if pause_on_false:
-                import pdbp
-                pdbp.set_trace()
-
-            if print_prompt_on_false:
-                print(prompt)
-
-            return False
-
-    return True
-
 def assert_before(
     child,
     patts: list[str],
 
-    **kwargs,
-
 ) -> None:
 
-    # as in before the prompt end
-    before: str = str(child.before.decode())
-    assert in_prompt_msg(
-        prompt=before,
-        parts=patts,
+    before = str(child.before.decode())
 
-        **kwargs
-    )
+    for patt in patts:
+        try:
+            assert patt in before
+        except AssertionError:
+            print(before)
+            raise
 
 
 @pytest.fixture(
@@ -232,10 +195,7 @@ def test_root_actor_error(spawn, user_in_out):
     before = str(child.before.decode())
 
     # make sure expected logging and error arrives
-    assert in_prompt_msg(
-        before,
-        [_crash_msg, "('root'"]
-    )
+    assert "Attaching to pdb in crashed actor: ('root'" in before
     assert 'AssertionError' in before
 
     # send user command
@@ -372,10 +332,7 @@ def test_subactor_error(
     child.expect(PROMPT)
 
     before = str(child.before.decode())
-    assert in_prompt_msg(
-        before,
-        [_crash_msg, "('name_error'"]
-    )
+    assert "Attaching to pdb in crashed actor: ('name_error'" in before
 
     if do_next:
         child.sendline('n')
@@ -396,15 +353,9 @@ def test_subactor_error(
     before = str(child.before.decode())
 
     # root actor gets debugger engaged
-    assert in_prompt_msg(
-        before,
-        [_crash_msg, "('root'"]
-    )
+    assert "Attaching to pdb in crashed actor: ('root'" in before
     # error is a remote error propagated from the subactor
-    assert in_prompt_msg(
-        before,
-        [_crash_msg, "('name_error'"]
-    )
+    assert "RemoteActorError: ('name_error'" in before
 
     # another round
     if ctlc:
@@ -429,10 +380,7 @@ def test_subactor_breakpoint(
     child.expect(PROMPT)
 
     before = str(child.before.decode())
-    assert in_prompt_msg(
-        before,
-        [_pause_msg, "('breakpoint_forever'"]
-    )
+    assert "Attaching pdb to actor: ('breakpoint_forever'" in before
 
     # do some "next" commands to demonstrate recurrent breakpoint
     # entries
@@ -448,10 +396,7 @@ def test_subactor_breakpoint(
         child.sendline('continue')
         child.expect(PROMPT)
         before = str(child.before.decode())
-        assert in_prompt_msg(
-            before,
-            [_pause_msg, "('breakpoint_forever'"]
-        )
+        assert "Attaching pdb to actor: ('breakpoint_forever'" in before
 
         if ctlc:
             do_ctlc(child)
@@ -496,10 +441,7 @@ def test_multi_subactors(
     child.expect(PROMPT)
 
     before = str(child.before.decode())
-    assert in_prompt_msg(
-        before,
-        [_pause_msg, "('breakpoint_forever'"]
-    )
+    assert "Attaching pdb to actor: ('breakpoint_forever'" in before
 
     if ctlc:
         do_ctlc(child)
@@ -519,10 +461,7 @@ def test_multi_subactors(
     # first name_error failure
     child.expect(PROMPT)
     before = str(child.before.decode())
-    assert in_prompt_msg(
-        before,
-        [_crash_msg, "('name_error'"]
-    )
+    assert "Attaching to pdb in crashed actor: ('name_error'" in before
     assert "NameError" in before
 
     if ctlc:
@@ -548,10 +487,7 @@ def test_multi_subactors(
     child.sendline('c')
     child.expect(PROMPT)
     before = str(child.before.decode())
-    assert in_prompt_msg(
-        before,
-        [_pause_msg, "('breakpoint_forever'"]
-    )
+    assert "Attaching pdb to actor: ('breakpoint_forever'" in before
 
     if ctlc:
         do_ctlc(child)
@@ -591,12 +527,9 @@ def test_multi_subactors(
     child.expect(PROMPT)
     before = str(child.before.decode())
 
-    assert_before(
-        child, [
+    assert_before(child, [
         # debugger attaches to root
-            # "Attaching to pdb in crashed actor: ('root'",
-            _crash_msg,
-            "('root'",
+        "Attaching to pdb in crashed actor: ('root'",
 
         # expect a multierror with exceptions for each sub-actor
         "RemoteActorError: ('breakpoint_forever'",
@@ -604,8 +537,7 @@ def test_multi_subactors(
         "RemoteActorError: ('spawn_error'",
         "RemoteActorError: ('name_error_1'",
         'bdb.BdbQuit',
-        ]
-    )
+    ])
 
     if ctlc:
         do_ctlc(child)
@@ -642,22 +574,15 @@ def test_multi_daemon_subactors(
     # the root's tty lock first so anticipate either crash
     # message on the first entry.
 
-    bp_forev_parts = [_pause_msg, "('bp_forever'"]
-    bp_forev_in_msg = partial(
-        in_prompt_msg,
-        parts=bp_forev_parts,
-    )
-
+    bp_forever_msg = "Attaching pdb to actor: ('bp_forever'"
     name_error_msg = "NameError: name 'doggypants' is not defined"
-    name_error_parts = [name_error_msg]
 
     before = str(child.before.decode())
-
-    if bp_forev_in_msg(prompt=before):
-        next_parts = name_error_parts
+    if bp_forever_msg in before:
+        next_msg = name_error_msg
 
     elif name_error_msg in before:
-        next_parts = bp_forev_parts
+        next_msg = bp_forever_msg
 
     else:
         raise ValueError("Neither log msg was found !?")
@@ -674,10 +599,7 @@ def test_multi_daemon_subactors(
 
     child.sendline('c')
     child.expect(PROMPT)
-    assert_before(
-        child,
-        next_parts,
-    )
+    assert_before(child, [next_msg])
 
     # XXX: hooray the root clobbering the child here was fixed!
     # IMO, this demonstrates the true power of SC system design.
@@ -701,15 +623,9 @@ def test_multi_daemon_subactors(
     child.expect(PROMPT)
 
     try:
-        assert_before(
-            child,
-            bp_forev_parts,
-        )
+        assert_before(child, [bp_forever_msg])
     except AssertionError:
-        assert_before(
-            child,
-            name_error_parts,
-        )
+        assert_before(child, [name_error_msg])
 
     else:
         if ctlc:
@@ -721,10 +637,7 @@ def test_multi_daemon_subactors(
 
         child.sendline('c')
         child.expect(PROMPT)
-        assert_before(
-            child,
-            name_error_parts,
-        )
+        assert_before(child, [name_error_msg])
 
     # wait for final error in root
     # where it crashs with boxed error
@@ -734,7 +647,7 @@ def test_multi_daemon_subactors(
             child.expect(PROMPT)
             assert_before(
                 child,
-                bp_forev_parts
+                [bp_forever_msg]
             )
         except AssertionError:
             break
@@ -743,9 +656,7 @@ def test_multi_daemon_subactors(
         child,
         [
             # boxed error raised in root task
-            # "Attaching to pdb in crashed actor: ('root'",
-            _crash_msg,
-            "('root'",
+            "Attaching to pdb in crashed actor: ('root'",
             "_exceptions.RemoteActorError: ('name_error'",
         ]
     )
@@ -859,7 +770,7 @@ def test_multi_nested_subactors_error_through_nurseries(
 
     child = spawn('multi_nested_subactors_error_up_through_nurseries')
 
-    # timed_out_early: bool = False
+    timed_out_early: bool = False
 
     for send_char in itertools.cycle(['c', 'q']):
         try:
@@ -960,14 +871,11 @@ def test_root_nursery_cancels_before_child_releases_tty_lock(
 
     if not timed_out_early:
         before = str(child.before.decode())
-        assert_before(
-            child,
-            [
+        assert_before(child, [
             "tractor._exceptions.RemoteActorError: ('spawner0'",
             "tractor._exceptions.RemoteActorError: ('name_error'",
             "NameError: name 'doggypants' is not defined",
-            ],
-        )
+        ])
 
 
 def test_root_cancels_child_context_during_startup(
@@ -1001,10 +909,8 @@ def test_different_debug_mode_per_actor(
 
     # only one actor should enter the debugger
     before = str(child.before.decode())
-    assert in_prompt_msg(
-        before,
-        [_crash_msg, "('debugged_boi'", "RuntimeError"],
-    )
+    assert "Attaching to pdb in crashed actor: ('debugged_boi'" in before
+    assert "RuntimeError" in before
 
     if ctlc:
         do_ctlc(child)

tests/test_rpc.py

@@ -134,7 +134,7 @@ def test_rpc_errors(
         value = err.value
 
         # might get multiple `trio.Cancelled`s as well inside an inception
-        if isinstance(value, ExceptionGroup):
+        if isinstance(value, trio.MultiError):
             value = next(itertools.dropwhile(
                 lambda exc: not isinstance(exc, tractor.RemoteActorError),
                 value.exceptions

tractor/_entry.py

@@ -116,18 +116,14 @@ def _trio_main(
     if actor.loglevel is not None:
         get_console_log(actor.loglevel)
         import os
-        actor_info: str = (
+        log.info(
+            'Started new trio process:\n'
             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(
-            'Started new trio process:\n'
-            +
-            actor_info
-        )
 
     try:
         if infect_asyncio:
@@ -137,14 +133,8 @@ def _trio_main(
             trio.run(trio_main)
     except KeyboardInterrupt:
         log.cancel(
-            'Actor received KBI\n'
-            +
-            actor_info
+            f'@{actor.uid} received KBI'
         )
 
     finally:
-        log.info(
-            'Actor terminated\n'
-            +
-            actor_info
-        )
+        log.info(f"Actor {actor.uid} terminated")

tractor/_portal.py

@@ -257,7 +257,7 @@ class Portal:
             return False
 
         reminfo: str = (
-            f'`Portal.cancel_actor()` => {self.channel.uid}\n'
+            f'{self.channel.uid}\n'
             f' |_{chan}\n'
         )
         log.cancel(
@@ -949,13 +949,9 @@ class Portal:
 
                 # CASE 1
                 else:
-                    outcome_str: str = ctx.repr_outcome(
-                        show_error_fields=True,
-                        # type_only=True,
-                    )
                     log.cancel(
-                        f'Context terminated due to local scope error:\n\n'
-                        f'{ctx.chan.uid} => {outcome_str}\n'
+                        f'Context terminated due to local scope error:\n'
+                        f'{etype.__name__}\n'
                     )
 
             # FINALLY, remove the context from runtime tracking and

tractor/_root.py

@@ -342,9 +342,7 @@ async def open_root_actor(
                 #     for an in nurseries:
                 #         tempn.start_soon(an.exited.wait)
 
-                logger.info(
-                    'Closing down root actor'
-                )
+                logger.cancel("Shutting down root actor")
                 await actor.cancel(None)  # self cancel
     finally:
         _state._current_actor = None

tractor/_runtime.py

@@ -627,7 +627,7 @@ async def _invoke(
                         # -[ ] textwrap.indent() it correctly!
                         # -[ ] BUT we need to wait until
                         #   the state is filled out before emitting
-                        #   this msg right ow its kinda empty? bleh..
+                        #   this msg right?
                         #
                         # f'  |_{ctx}'
                     )
@@ -638,6 +638,7 @@ async def _invoke(
                         msg += (
                             # '------ - ------\n'
                             # 'IPC msg:\n'
+                            # f'\n{ctx._cancel_msg}'
                             f'\n\n{ctx._cancel_msg}'
                         )
 
@@ -672,13 +673,28 @@ async def _invoke(
             # if a remote error was set then likely the
             # exception group was raised due to that, so
             # and we instead raise that error immediately!
+            # if (
+            #     # ctx.side == 'caller'
+            #     # ctx._portal
+            #     # and (re := ctx._remote_error)
+            #     re := ctx._remote_error
+            # ):
+            # if re := ctx._remote_error:
+            #     ctx._maybe_raise_remote_err(re)
             ctx.maybe_raise()
 
-            # maybe TODO: pack in come kinda
-            # `trio.Cancelled.__traceback__` here so they can be
-            # unwrapped and displayed on the caller side? no se..
+            # if isinstance(scope_error, KeyboardInterrupt):
+            #     await pause()
+
+            # maybe TODO: pack in
+            # ``trio.Cancelled.__traceback__`` here so they can
+            # be unwrapped and displayed on the caller side?
             raise
 
+        # except BaseException as scope_error:
+        #     ctx._local_error = 
+        # else:
+
         # `@context` entrypoint task bookeeping.
         # i.e. only pop the context tracking if used ;)
         finally:
@@ -691,6 +707,13 @@ async def _invoke(
                 (chan.uid, cid)
             )
 
+            #     ctx.maybe_error
+            #     # ctx._local_error
+            #     # or ctx._remote_error
+            # )
+            # if not final_err and ctx.canceller:
+            #     await pause()
+
             merr: Exception|None = ctx.maybe_error
 
             (
@@ -1094,6 +1117,7 @@ class Actor:
         # append new channel
         # TODO: can we just use list-ref directly?
         chans.append(chan)
+        # self._peers[uid].append(chan)
 
         log.runtime(con_msg)
 
@@ -1224,6 +1248,7 @@ class Actor:
                 f'uid: {chan.uid}\n'
                 f'|_{pformat(chan)}\n'
             )
+            # chans = self._peers.get(chan.uid)
             chans.remove(chan)
 
             # TODO: do we need to be this pedantic?

tractor/devx/__init__.py

@@ -21,17 +21,30 @@ and working with/on the actor runtime.
 
 """
 from ._debug import (
-    maybe_wait_for_debugger as maybe_wait_for_debugger,
-    acquire_debug_lock as acquire_debug_lock,
-    breakpoint as breakpoint,
-    pause as pause,
-    pause_from_sync as pause_from_sync,
-    shield_sigint_handler as shield_sigint_handler,
-    MultiActorPdb as MultiActorPdb,
-    open_crash_handler as open_crash_handler,
-    maybe_open_crash_handler as maybe_open_crash_handler,
-    post_mortem as post_mortem,
+    maybe_wait_for_debugger,
+    acquire_debug_lock,
+    breakpoint,
+    pause,
+    pause_from_sync,
+    shield_sigint_handler,
+    MultiActorPdb,
+    open_crash_handler,
+    maybe_open_crash_handler,
+    post_mortem,
 )
 from ._stackscope import (
     enable_stack_on_sig as enable_stack_on_sig,
 )
+
+__all__ = [
+    'maybe_wait_for_debugger',
+    'acquire_debug_lock',
+    'breakpoint',
+    'pause',
+    'pause_from_sync',
+    'shield_sigint_handler',
+    'MultiActorPdb',
+    'open_crash_handler',
+    'maybe_open_crash_handler',
+    'post_mortem',
+]

tractor/devx/_debug.py

@@ -21,19 +21,18 @@ Multi-core debugging for da peeps!
 """
 from __future__ import annotations
 import bdb
+import os
+import sys
+import signal
+from functools import (
+    partial,
+    cached_property,
+)
 from contextlib import (
     asynccontextmanager as acm,
     contextmanager as cm,
     nullcontext,
 )
-from functools import (
-    partial,
-    cached_property,
-)
-import os
-import signal
-import sys
-import traceback
 from typing import (
     Any,
     Callable,
@@ -612,9 +611,6 @@ def shield_sigint_handler(
         # https://github.com/prompt-toolkit/python-prompt-toolkit/blob/c2c6af8a0308f9e5d7c0e28cb8a02963fe0ce07a/prompt_toolkit/patch_stdout.py
 
 
-_pause_msg: str = 'Attaching to pdb REPL in actor'
-
-
 def _set_trace(
     actor: tractor.Actor | None = None,
     pdb: MultiActorPdb | None = None,
@@ -636,13 +632,7 @@ def _set_trace(
         ) or shield
     ):
         # pdbp.set_trace()
-        # TODO: maybe print the actor supervion tree up to the
-        # root here? Bo
-        log.pdb(
-            f'{_pause_msg}\n'
-            '|\n'
-            f'|_ {actor.uid}\n'
-        )
+        log.pdb(f"\nAttaching pdb to actor: {actor.uid}\n")
         # no f!#$&* idea, but when we're in async land
         # we need 2x frames up?
         frame = frame.f_back
@@ -921,11 +911,6 @@ async def breakpoint(**kwargs):
     await pause(**kwargs)
 
 
-_crash_msg: str = (
-    'Attaching to pdb REPL in crashed actor'
-)
-
-
 def _post_mortem(
     actor: tractor.Actor,
     pdb: MultiActorPdb,
@@ -936,23 +921,15 @@ def _post_mortem(
     debugger instance.
 
     '''
-    # TODO: print the actor supervion tree up to the root
-    # here! Bo
-    log.pdb(
-        f'{_crash_msg}\n'
-        '|\n'
-        f'|_ {actor.uid}\n'
-    )
+    log.pdb(f"\nAttaching to pdb in crashed actor: {actor.uid}\n")
 
-    # TODO: only replacing this to add the
-    # `end=''` to the print XD
-    # pdbp.xpm(Pdb=lambda: pdb)
-    info = sys.exc_info()
-    print(traceback.format_exc(), end='')
-    pdbp.post_mortem(
-        t=info[2],
-        Pdb=lambda: pdb,
-    )
+    # TODO: you need ``pdbpp`` master (at least this commit
+    # https://github.com/pdbpp/pdbpp/commit/b757794857f98d53e3ebbe70879663d7d843a6c2)
+    # to fix this and avoid the hang it causes. See issue:
+    # https://github.com/pdbpp/pdbpp/issues/480
+    # TODO: help with a 3.10+ major release if/when it arrives.
+
+    pdbp.xpm(Pdb=lambda: pdb)
 
 
 post_mortem = partial(
@@ -1024,13 +1001,13 @@ async def maybe_wait_for_debugger(
 
     header_msg: str = '',
 
-) -> bool:  # was locked and we polled?
+) -> None:
 
     if (
         not debug_mode()
         and not child_in_debug
     ):
-        return False
+        return
 
 
     msg: str = header_msg
@@ -1048,7 +1025,8 @@ async def maybe_wait_for_debugger(
 
         if sub_in_debug := Lock.global_actor_in_debug:
             msg += (
-                f'Debug `Lock` in use by subactor: {sub_in_debug}\n'
+                'Debug `Lock` in use by subactor\n'
+                f'|_{sub_in_debug}\n'
             )
             # TODO: could this make things more deterministic?
             # wait to see if a sub-actor task will be
@@ -1057,12 +1035,12 @@ async def maybe_wait_for_debugger(
             # XXX => but it doesn't seem to work..
             # await trio.testing.wait_all_tasks_blocked(cushion=0)
         else:
-            log.debug(
+            log.pdb(
                 msg
                 +
                 'Root immediately acquired debug TTY LOCK'
             )
-            return False
+            return
 
         for istep in range(poll_steps):
 
@@ -1112,13 +1090,12 @@ async def maybe_wait_for_debugger(
                     continue
 
         # fallthrough on failure to acquire..
-        # else:
-        #     raise RuntimeError(
-        #         msg
-        #         +
-        #         'Root actor failed to acquire debug lock?'
-        #     )
-        return True
+        else:
+            raise RuntimeError(
+                msg
+                +
+                'Root actor failed to acquire debug lock?'
+            )
 
     # else:
     #     # TODO: non-root call for #320?
@@ -1127,7 +1104,6 @@ async def maybe_wait_for_debugger(
     #         subactor_uid=this_uid,
     #     ):
     #         pass
-    return False
 
 # TODO: better naming and what additionals?
 # - [ ] optional runtime plugging?

tractor/devx/cli.py

@@ -23,6 +23,10 @@ Currently popular frameworks supported are:
 
 """
 from __future__ import annotations
+from contextlib import (
+    # asynccontextmanager as acm,
+    contextmanager as cm,
+)
 from typing import (
     Any,
     Callable,
@@ -32,6 +36,9 @@ from typing_extensions import Annotated
 import typer
 
 
+from ._debug import open_crash_handler
+
+
 _runtime_vars: dict[str, Any] = {}