TOSQASH 22e62ed: with-stmt-ws-removal

Nowhere near ready yet since this generates many test-body-logic
un-handled true-positives which currently fail, but it's a first draft
for the general case set. To start, includes a greater-than-one-`Msg` (a
strict top level tagged-union-of-structs) pld spec alongside
a `AnyFieldMsg`-workaround struct-msg for packing all other builtin
non-`Struct`/`Any` python types alongside the other explicit msgs.

.claude/settings.local.json

@@ -0,0 +1,15 @@
+{
+  "permissions": {
+    "allow": [
+      "Write(.claude/*commit_msg*)",
+      "Write(.claude/git_commit_msg_LATEST.md)",
+      "Bash(date *)",
+      "Bash(cp .claude/*)",
+      "Bash(git diff *)",
+      "Bash(git log *)",
+      "Bash(git status)"
+    ],
+    "deny": [],
+    "ask": []
+  }
+}

.claude/skills/commit-msg/SKILL.md

@@ -0,0 +1,86 @@
+---
+name: commit-msg
+description: >
+  Generate git commit messages following project style. Use when user
+  wants to create a commit or asks for a commit message.
+argument-hint: "[optional-scope-or-description]"
+disable-model-invocation: true
+allowed-tools:
+  - Bash(git *)
+  - Bash(date *)
+  - Bash(cp *)
+  - Read
+  - Grep
+  - Glob
+  - Write
+---
+
+When generating commit messages, always follow this process:
+
+0. **Check for staged changes**: if `git diff --staged` is
+   empty, STOP and tell the user "nothing is staged!" with
+   a reminder to `git add` before invoking this skill.
+
+1. **Gather context** from the staged diff and recent
+   history:
+
+   - Staged changes: !`git diff --staged --stat`
+   - Recent commits: !`git log --oneline -5`
+
+2. **Analyze the diff**: understand what files changed and
+   the nature of the changes (new feature, bug fix, refactor,
+   etc.)
+
+3. **Write the commit message** following these rules:
+
+**Use the accompanying style guide:**
+- See [style-guide-reference.md](style-guide-reference.md)
+  for detailed analysis of past commits in this repo.
+
+**Subject Line Format:**
+- Present tense verbs: Add, Drop, Fix, Use, Move, Adjust, etc.
+- Target 50 chars (hard max: 67)
+- Backticks around ALL code elements (classes, functions, modules, vars)
+- Specific about what changed
+
+**Body Format (optional - keep simple if warranted):**
+- Max 67 char line length
+- Use `-` bullets for lists
+- Section markers: `Also,` `Deats,` `Other,` `Further,`
+- Abbreviations: msg, bg, ctx, impl, mod, obvi, tn, fn, bc, var, prolly
+- Casual yet technically precise tone
+- Never write lines with only whitespace
+
+**Common Opening Patterns:**
+- New features: "Add `feature` to `module`"
+- Removals: "Drop `attr` from `class`"
+- Bug fixes: "Fix `issue` in `function`"
+- Code moves: "Move `thing` to `new_location`"
+- Adoption: "Use `new_tool` for `task`"
+- Minor tweaks: "Adjust `behavior` in `component`"
+
+4. **Write to TWO files**:
+   - `.claude/<timestamp>_<hash>_commit_msg.md`
+     * with `<timestamp>` from `date -u +%Y%m%dT%H%M%SZ` or similar
+       filesystem-safe format.
+     * and `<hash>` from `git log -1 --format=%h` first 7 chars.
+   - `.claude/git_commit_msg_LATEST.md` (overwrite)
+
+5. **Always include credit footer**:
+
+When no part of the patch was written by `claude`,
+```
+(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
+[claude-code-gh]: https://github.com/anthropics/claude-code
+```
+
+when some or all of the patch was written by `claude`
+instead use,
+
+```
+(this patch was generated in some part by [`claude-code`][claude-code-gh])
+[claude-code-gh]: https://github.com/anthropics/claude-code
+```
+
+Keep it concise. Match the tone of recent commits. For simple
+changes, use subject line only.

.claude/skills/commit-msg/style-guide-reference.md

@@ -0,0 +1,225 @@
+# Commit Message Style Guide for `tractor`
+
+Analysis based on 500 recent commits from the `tractor` repository.
+
+## Core Principles
+
+Write commit messages that are technically precise yet casual in
+tone. Use abbreviations and informal language while maintaining
+clarity about what changed and why.
+
+## Subject Line Format
+
+### Length and Structure
+- Target: ~50 chars with a hard-max of 67.
+- Use backticks around code elements (72.2% of commits)
+- Rarely use colons (5.2%), except for file prefixes
+- End with '?' for uncertain changes (rare: 0.8%)
+- End with '!' for important changes (rare: 2.0%)
+
+### Opening Verbs (Present Tense)
+
+Most common verbs from analysis:
+- `Add` (14.4%) - wholly new features/functionality
+- `Use` (4.4%) - adopt new approach/tool
+- `Drop` (3.6%) - remove code/feature
+- `Fix` (2.4%) - bug fixes
+- `Move`/`Mv` (3.6%) - relocate code
+- `Adjust` (2.0%) - minor tweaks
+- `Update` (1.6%) - enhance existing feature
+- `Bump` (1.2%) - dependency updates
+- `Rename` (1.2%) - identifier changes
+- `Set` (1.2%) - configuration changes
+- `Handle` (1.0%) - add handling logic
+- `Raise` (1.0%) - add error raising
+- `Pass` (0.8%) - pass parameters/values
+- `Support` (0.8%) - add support for something
+- `Hide` (1.4%) - make private/internal
+- `Always` (1.4%) - enforce consistent behavior
+- `Mk` (1.4%) - make/create (abbreviated)
+- `Start` (1.0%) - begin implementation
+
+Other frequent verbs: `More`, `Change`, `Extend`, `Disable`, `Log`,
+`Enable`, `Ensure`, `Expose`, `Allow`
+
+### Backtick Usage
+
+Always use backticks for:
+- Module names: `trio`, `asyncio`, `msgspec`, `greenback`, `stackscope`
+- Class names: `Context`, `Actor`, `Address`, `PldRx`, `SpawnSpec`
+- Method names: `.pause_from_sync()`, `._pause()`, `.cancel()`
+- Function names: `breakpoint()`, `collapse_eg()`, `open_root_actor()`
+- Decorators: `@acm`, `@context`
+- Exceptions: `Cancelled`, `TransportClosed`, `MsgTypeError`
+- Keywords: `finally`, `None`, `False`
+- Variable names: `tn`, `debug_mode`
+- Complex expressions: `trio.Cancelled`, `asyncio.Task`
+
+Most backticked terms in tractor:
+`trio`, `asyncio`, `Context`, `.pause_from_sync()`, `tn`,
+`._pause()`, `breakpoint()`, `collapse_eg()`, `Actor`, `@acm`,
+`.cancel()`, `Cancelled`, `open_root_actor()`, `greenback`
+
+### Examples
+
+Good subject lines:
+```
+Add `uds` to `._multiaddr`, tweak typing
+Drop `DebugStatus.shield` attr, add `.req_finished`
+Use `stackscope` for all actor-tree rendered "views"
+Fix `.to_asyncio` inter-task-cancellation!
+Bump `ruff.toml` to target py313
+Mv `load_module_from_path()` to new `._code_load` submod
+Always use `tuple`-cast for singleton parent addrs
+```
+
+## Body Format
+
+### General Structure
+- 43.2% of commits have no body (simple changes)
+- Use blank line after subject
+- Max line length: 67 chars
+- Use `-` bullets for lists (28.0% of commits)
+- Rarely use `*` bullets (2.4%)
+
+### Section Markers
+
+Use these markers to organize longer commit bodies:
+- `Also,` (most common: 26 occurrences)
+- `Other,` (13 occurrences)
+- `Deats,` (11 occurrences) - for implementation details
+- `Further,` (7 occurrences)
+- `TODO,` (3 occurrences)
+- `Impl details,` (2 occurrences)
+- `Notes,` (1 occurrence)
+
+### Common Abbreviations
+
+Use these freely (sorted by frequency):
+- `msg` (63) - message
+- `bg` (37) - background
+- `ctx` (30) - context
+- `impl` (27) - implementation
+- `mod` (26) - module
+- `obvi` (17) - obviously
+- `tn` (16) - task name
+- `fn` (15) - function
+- `vs` (15) - versus
+- `bc` (14) - because
+- `var` (14) - variable
+- `prolly` (9) - probably
+- `ep` (6) - entry point
+- `OW` (5) - otherwise
+- `rn` (4) - right now
+- `sig` (4) - signal/signature
+- `deps` (3) - dependencies
+- `iface` (2) - interface
+- `subproc` (2) - subprocess
+- `tho` (2) - though
+- `ofc` (2) - of course
+
+### Tone and Style
+
+- Casual but technical (use `XD` for humor: 23 times)
+- Use `..` for trailing thoughts (108 occurrences)
+- Use `Woops,` to acknowledge mistakes (4 subject lines)
+- Don't be afraid to show personality while being precise
+
+### Example Bodies
+
+Simple with bullets:
+```
+Add `multiaddr` and bump up some deps
+
+Since we're planning to use it for (discovery)
+addressing, allowing replacement of the hacky (pretend)
+attempt in `tractor._multiaddr` Bp
+
+Also pin some deps,
+- make us py312+
+- use `pdbp` with my frame indexing fix.
+- mv to latest `xonsh` for fancy cmd/suggestion injections.
+
+Bump lock file to match obvi!
+```
+
+With section markers:
+```
+Use `stackscope` for all actor-tree rendered "views"
+
+Instead of the (much more) limited and hacky `.devx._code`
+impls, move to using the new `.devx._stackscope` API which
+wraps the `stackscope` project.
+
+Deats,
+- make new `stackscope.extract_stack()` wrapper
+- port over frame-descing to `_stackscope.pformat_stack()`
+- move `PdbREPL` to use `stackscope` render approach
+- update tests for new stack output format
+
+Also,
+- tweak log formatting for consistency
+- add typing hints throughout
+```
+
+## Special Patterns
+
+### WIP Commits
+Rare (0.2%) - avoid committing WIP if possible
+
+### Merge Commits
+Auto-generated (4.4%), don't worry about style
+
+### File References
+- Use `module.py` or `.submodule` style
+- Rarely use `file.py:line` references (0 in analysis)
+
+### Links
+- GitHub links used sparingly (3 total)
+- Prefer code references over external links
+
+## Footer
+
+The default footer should credit `claude` (you) for helping generate
+the commit msg content:
+
+```
+(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
+[claude-code-gh]: https://github.com/anthropics/claude-code
+```
+
+Further, if the patch was solely or in part written
+by `claude`, instead add:
+
+```
+(this patch was generated in some part by [`claude-code`][claude-code-gh])
+[claude-code-gh]: https://github.com/anthropics/claude-code
+```
+
+## Summary Checklist
+
+Before committing, verify:
+- [ ] Subject line uses present tense verb
+- [ ] Subject line ~50 chars (hard max 67)
+- [ ] Code elements wrapped in backticks
+- [ ] Body lines ≤67 chars
+- [ ] Abbreviations used where natural
+- [ ] Casual yet precise tone
+- [ ] Section markers if body >3 paragraphs
+- [ ] Technical accuracy maintained
+
+## Analysis Metadata
+
+```
+Source: tractor repository
+Commits analyzed: 500
+Date range: 2019-2025
+Analysis date: 2026-02-08
+```
+
+---
+
+(this style guide was generated by [`claude-code`][claude-code-gh]
+analyzing commit history)
+
+[claude-code-gh]: https://github.com/anthropics/claude-code

.github/workflows/ci.yml

@@ -74,24 +74,44 @@ jobs:
   #       run: mypy tractor/ --ignore-missing-imports --show-traceback
 
 
-  testing-linux:
-    name: '${{ matrix.os }} Python ${{ matrix.python }} - ${{ matrix.spawn_backend }}'
-    timeout-minutes: 10
+  testing:
+    name: '${{ matrix.os }} Python${{ matrix.python-version }} spawn_backend=${{ matrix.spawn_backend }} tpt_proto=${{ matrix.tpt_proto }}'
+    timeout-minutes: 16
     runs-on: ${{ matrix.os }}
 
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-latest]
-        python-version: ['3.13']
+        os: [
+          ubuntu-latest,
+          macos-latest,
+        ]
+        python-version: [
+          '3.13',
+          # '3.14',
+        ]
         spawn_backend: [
           'trio',
           # 'mp_spawn',
           # 'mp_forkserver',
+          # ?TODO^ is it worth it to get these running again?
+          #
+          # - [ ] next-gen backends, on 3.13+
+          #   https://github.com/goodboy/tractor/issues/379
+          # 'subinterpreter',
+          # 'subint',
         ]
+        tpt_proto: [
+          'tcp',
+          'uds',
+        ]
+        # https://github.com/orgs/community/discussions/26253#discussioncomment-3250989
+        exclude:
+          # don't do UDS run on macOS (for now)
+          - os: macos-latest
+            tpt_proto: 'uds'
 
     steps:
-
       - uses: actions/checkout@v4
 
       - name: 'Install uv + py-${{ matrix.python-version }}'
@@ -118,7 +138,11 @@ jobs:
         run: uv tree
 
       - name: Run tests
-        run: uv run pytest tests/ --spawn-backend=${{ matrix.spawn_backend }} -rsx
+        run: >
+          uv run
+          pytest tests/ -rsx
+          --spawn-backend=${{ matrix.spawn_backend }}
+          --tpt-proto=${{ matrix.tpt_proto }}
 
   # XXX legacy NOTE XXX
   #

.gitignore

@@ -102,3 +102,30 @@ venv.bak/
 
 # mypy
 .mypy_cache/
+
+# all files under
+.git/
+
+# any commit-msg gen tmp files
+.claude/*_commit_*.md
+.claude/*_commit*.toml
+.claude/*_commit*.txt
+
+# nix develop --profile .nixdev
+.nixdev*
+
+# :Obsession .
+Session.vim
+# `gish` local `.md`-files
+# TODO? better all around automation!
+# -[ ] it'd be handy to also commit and sync with wtv git service?
+# -[ ] everything should be put under a `.gish/` no?
+gitea/
+gh/
+
+# ------ macOS ------
+# Finder metadata
+**/.DS_Store
+
+# LLM conversations that should remain private
+docs/conversations/

docs/README.rst

@@ -420,20 +420,17 @@ Check out our experimental system for `guest`_-mode controlled
 
 
     async def aio_echo_server(
-        to_trio: trio.MemorySendChannel,
-        from_trio: asyncio.Queue,
+        chan: tractor.to_asyncio.LinkedTaskChannel,
     ) -> None:
 
         # a first message must be sent **from** this ``asyncio``
         # task or the ``trio`` side will never unblock from
         # ``tractor.to_asyncio.open_channel_from():``
-        to_trio.send_nowait('start')
+        chan.started_nowait('start')
 
-        # XXX: this uses an ``from_trio: asyncio.Queue`` currently but we
-        # should probably offer something better.
         while True:
             # echo the msg back
-            to_trio.send_nowait(await from_trio.get())
+            chan.send_nowait(await chan.get())
             await asyncio.sleep(0)
 
 
@@ -445,7 +442,7 @@ Check out our experimental system for `guest`_-mode controlled
         # message.
         async with tractor.to_asyncio.open_channel_from(
             aio_echo_server,
-        ) as (first, chan):
+        ) as (chan, first):
 
             assert first == 'start'
             await ctx.started(first)
@@ -504,8 +501,10 @@ Yes, we spawn a python process, run ``asyncio``, start ``trio`` on the
 ``asyncio`` loop, then send commands to the ``trio`` scheduled tasks to
 tell ``asyncio`` tasks what to do XD
 
-We need help refining the `asyncio`-side channel API to be more
-`trio`-like. Feel free to sling your opinion in `#273`_!
+The ``asyncio``-side task receives a single
+``chan: LinkedTaskChannel`` handle providing a ``trio``-like
+API: ``.started_nowait()``, ``.send_nowait()``, ``.get()``
+and more. Feel free to sling your opinion in `#273`_!
 
 
 .. _#273: https://github.com/goodboy/tractor/issues/273
@@ -641,13 +640,15 @@ Help us push toward the future of distributed `Python`.
 - Typed capability-based (dialog) protocols ( see `#196
   <https://github.com/goodboy/tractor/issues/196>`_ with draft work
   started in `#311 <https://github.com/goodboy/tractor/pull/311>`_)
-- We **recently disabled CI-testing on windows** and need help getting
-  it running again! (see `#327
-  <https://github.com/goodboy/tractor/pull/327>`_). **We do have windows
-  support** (and have for quite a while) but since no active hacker
-  exists in the user-base to help test on that OS, for now we're not
-  actively maintaining testing due to the added hassle and general
-  latency..
+- **macOS is now officially supported** and tested in CI
+  alongside Linux!
+- We **recently disabled CI-testing on windows** and need
+  help getting it running again! (see `#327
+  <https://github.com/goodboy/tractor/pull/327>`_). **We do
+  have windows support** (and have for quite a while) but
+  since no active hacker exists in the user-base to help
+  test on that OS, for now we're not actively maintaining
+  testing due to the added hassle and general latency..
 
 
 Feel like saying hi?

examples/debugging/asyncio_bp.py

@@ -18,15 +18,14 @@ async def aio_sleep_forever():
 
 
 async def bp_then_error(
-    to_trio: trio.MemorySendChannel,
-    from_trio: asyncio.Queue,
+    chan: to_asyncio.LinkedTaskChannel,
 
     raise_after_bp: bool = True,
 
 ) -> None:
 
     # sync with `trio`-side (caller) task
-    to_trio.send_nowait('start')
+    chan.started_nowait('start')
 
     # NOTE: what happens here inside the hook needs some refinement..
     # => seems like it's still `.debug._set_trace()` but
@@ -60,7 +59,7 @@ async def trio_ctx(
         to_asyncio.open_channel_from(
             bp_then_error,
             # raise_after_bp=not bp_before_started,
-        ) as (first, chan),
+        ) as (chan, first),
 
         trio.open_nursery() as tn,
     ):

examples/debugging/shield_hang_in_sub.py

@@ -3,6 +3,7 @@ Verify we can dump a `stackscope` tree on a hang.
 
 '''
 import os
+import platform
 import signal
 
 import trio
@@ -31,13 +32,26 @@ async def main(
     from_test: bool = False,
 ) -> None:
 
+    if platform.system() != 'Darwin':
+        tpt = 'uds'
+    else:
+        # XXX, precisely we can't use pytest's tmp-path generation
+        # for tests.. apparently because:
+        #
+        # > The OSError: AF_UNIX path too long in macOS Python occurs
+        # > because the path to the Unix domain socket exceeds the
+        # > operating system's maximum path length limit (around 104
+        #
+        # WHICH IS just, wtf hillarious XD
+        tpt = 'tcp'
+
     async with (
         tractor.open_nursery(
             debug_mode=True,
             enable_stack_on_sig=True,
             # maybe_enable_greenback=False,
             loglevel='devx',
-            enable_transports=['uds'],
+            enable_transports=[tpt],
         ) as an,
     ):
         ptl: tractor.Portal  = await an.start_actor(

examples/debugging/subactor_bp_in_ctx.py

@@ -1,3 +1,5 @@
+import platform
+
 import tractor
 import trio
 
@@ -34,9 +36,22 @@ async def just_bp(
 
 async def main():
 
+    if platform.system() != 'Darwin':
+        tpt = 'uds'
+    else:
+        # XXX, precisely we can't use pytest's tmp-path generation
+        # for tests.. apparently because:
+        #
+        # > The OSError: AF_UNIX path too long in macOS Python occurs
+        # > because the path to the Unix domain socket exceeds the
+        # > operating system's maximum path length limit (around 104
+        #
+        # WHICH IS just, wtf hillarious XD
+        tpt = 'tcp'
+
     async with tractor.open_nursery(
         debug_mode=True,
-        enable_transports=['uds'],
+        enable_transports=[tpt],
         loglevel='devx',
     ) as n:
         p = await n.start_actor(

examples/full_fledged_streaming_service.py

@@ -90,7 +90,7 @@ async def main() -> list[int]:
     # yes, a nursery which spawns `trio`-"actors" B)
     an: ActorNursery
     async with tractor.open_nursery(
-        loglevel='cancel',
+        loglevel='error',
         # debug_mode=True,
     ) as an:
 
@@ -118,8 +118,10 @@ async def main() -> list[int]:
         cancelled: bool = await portal.cancel_actor()
         assert cancelled
 
-        print(f"STREAM TIME = {time.time() - start}")
-        print(f"STREAM + SPAWN TIME = {time.time() - pre_start}")
+        print(
+            f"STREAM TIME = {time.time() - start}\n"
+            f"STREAM + SPAWN TIME = {time.time() - pre_start}\n"
+        )
         assert result_stream == list(range(seed))
         return result_stream
 

examples/infected_asyncio_echo_server.py

@@ -11,21 +11,17 @@ import tractor
 
 
 async def aio_echo_server(
-    to_trio: trio.MemorySendChannel,
-    from_trio: asyncio.Queue,
-
+    chan: tractor.to_asyncio.LinkedTaskChannel,
 ) -> None:
 
     # a first message must be sent **from** this ``asyncio``
     # task or the ``trio`` side will never unblock from
     # ``tractor.to_asyncio.open_channel_from():``
-    to_trio.send_nowait('start')
+    chan.started_nowait('start')
 
-    # XXX: this uses an ``from_trio: asyncio.Queue`` currently but we
-    # should probably offer something better.
     while True:
         # echo the msg back
-        to_trio.send_nowait(await from_trio.get())
+        chan.send_nowait(await chan.get())
         await asyncio.sleep(0)
 
 
@@ -37,7 +33,7 @@ async def trio_to_aio_echo_server(
     # message.
     async with tractor.to_asyncio.open_channel_from(
         aio_echo_server,
-    ) as (first, chan):
+    ) as (chan, first):
 
         assert first == 'start'
         await ctx.started(first)

pyproject.toml

@@ -9,7 +9,7 @@ name = "tractor"
 version = "0.1.0a6dev0"
 description = 'structured concurrent `trio`-"actors"'
 authors = [{ name = "Tyler Goodlet", email = "goodboy_foss@protonmail.com" }]
-requires-python = ">= 3.12"
+requires-python = ">=3.12, <3.14"
 readme = "docs/README.rst"
 license = "AGPL-3.0-or-later"
 keywords = [
@@ -24,11 +24,13 @@ keywords = [
 classifiers = [
   "Development Status :: 3 - Alpha",
   "Operating System :: POSIX :: Linux",
+  "Operating System :: MacOS",
   "Framework :: Trio",
   "License :: OSI Approved :: GNU Affero General Public License v3 or later (AGPLv3+)",
   "Programming Language :: Python :: Implementation :: CPython",
   "Programming Language :: Python :: 3 :: Only",
-  "Programming Language :: Python :: 3.11",
+  "Programming Language :: Python :: 3.12",
+  "Programming Language :: Python :: 3.13",
   "Topic :: System :: Distributed Computing",
 ]
 dependencies = [
@@ -162,6 +164,7 @@ all_bullets = true
 
 [tool.pytest.ini_options]
 minversion = '6.0'
+# https://docs.pytest.org/en/stable/reference/reference.html#configuration-options
 testpaths = [
   'tests'
 ]
@@ -172,10 +175,17 @@ addopts = [
   '--import-mode=importlib',
   # don't show frickin captured logs AGAIN in the report..
   '--show-capture=no',
+
+  # disable `xonsh` plugin
+  # https://docs.pytest.org/en/stable/how-to/plugins.html#disabling-plugins-from-autoloading
+  # https://docs.pytest.org/en/stable/how-to/plugins.html#deactivating-unregistering-a-plugin-by-name
+  '-p no:xonsh'
 ]
 log_cli = false
 # TODO: maybe some of these layout choices?
 # https://docs.pytest.org/en/8.0.x/explanation/goodpractices.html#choosing-a-test-layout-import-rules
 # pythonpath = "src"
 
+# https://docs.pytest.org/en/stable/reference/reference.html#confval-console_output_style
+console_output_style = 'progress'
 # ------ tool.pytest ------

pytest.ini

@@ -1,8 +0,0 @@
-# vim: ft=ini
-# pytest.ini for tractor
-
-[pytest]
-# don't show frickin captured logs AGAIN in the report..
-addopts = --show-capture='no'
-log_cli = false
-; minversion = 6.0

tests/conftest.py

@@ -11,6 +11,7 @@ import platform
 import time
 
 import pytest
+import tractor
 from tractor._testing import (
     examples_dir as examples_dir,
     tractor_test as tractor_test,
@@ -22,6 +23,8 @@ pytest_plugins: list[str] = [
     'tractor._testing.pytest',
 ]
 
+_ci_env: bool = os.environ.get('CI', False)
+_non_linux: bool = platform.system() != 'Linux'
 
 # Sending signal.SIGINT on subprocess fails on windows. Use CTRL_* alternatives
 if platform.system() == 'Windows':
@@ -34,9 +37,8 @@ else:
     _INT_SIGNAL = signal.SIGINT
     _INT_RETURN_CODE = 1 if sys.version_info < (3, 8) else -signal.SIGINT.value
     _PROC_SPAWN_WAIT = (
-        0.6
-        if sys.version_info < (3, 7)
-        else 0.4
+        2 if _ci_env
+        else 1
     )
 
 
@@ -44,6 +46,10 @@ no_windows = pytest.mark.skipif(
     platform.system() == "Windows",
     reason="Test is unsupported on windows",
 )
+no_macos = pytest.mark.skipif(
+    platform.system() == "Darwin",
+    reason="Test is unsupported on MacOS",
+)
 
 
 def pytest_addoption(
@@ -61,7 +67,7 @@ def pytest_addoption(
 
 
 @pytest.fixture(scope='session', autouse=True)
-def loglevel(request):
+def loglevel(request) -> str:
     import tractor
     orig = tractor.log._default_loglevel
     level = tractor.log._default_loglevel = request.config.option.loglevel
@@ -69,12 +75,44 @@ def loglevel(request):
         level=level,
         name='tractor',  # <- enable root logger
     )
-    log.info(f'Test-harness logging level: {level}\n')
+    log.info(
+        f'Test-harness set runtime loglevel: {level!r}\n'
+    )
     yield level
     tractor.log._default_loglevel = orig
 
 
-_ci_env: bool = os.environ.get('CI', False)
+@pytest.fixture(scope='function')
+def test_log(
+    request,
+    loglevel: str,
+) -> tractor.log.StackLevelAdapter:
+    '''
+    Deliver a per test-module-fn logger instance for reporting from
+    within actual test bodies/fixtures.
+
+    For example this can be handy to report certain error cases from
+    exception handlers using `test_log.exception()`.
+
+    '''
+    modname: str = request.function.__module__
+    log = tractor.log.get_logger(
+        name=modname,  # <- enable root logger
+        # pkg_name='tests',
+    )
+    _log = tractor.log.get_console_log(
+        level=loglevel,
+        logger=log,
+        name=modname,
+        # pkg_name='tests',
+    )
+    _log.debug(
+        f'In-test-logging requested\n'
+        f'test_log.name: {log.name!r}\n'
+        f'level: {loglevel!r}\n'
+
+    )
+    yield _log
 
 
 @pytest.fixture(scope='session')
@@ -110,6 +148,8 @@ def daemon(
     testdir: pytest.Pytester,
     reg_addr: tuple[str, int],
     tpt_proto: str,
+    ci_env: bool,
+    test_log: tractor.log.StackLevelAdapter,
 
 ) -> subprocess.Popen:
     '''
@@ -125,10 +165,12 @@ def daemon(
         "import tractor; "
         "tractor.run_daemon([], "
         "registry_addrs={reg_addrs}, "
+        "enable_transports={enable_tpts}, "
         "debug_mode={debug_mode}, "
         "loglevel={ll})"
     ).format(
         reg_addrs=str([reg_addr]),
+        enable_tpts=str([tpt_proto]),
         ll="'{}'".format(loglevel) if loglevel else None,
         debug_mode=debug_mode,
     )
@@ -147,13 +189,25 @@ def daemon(
         **kwargs,
     )
 
+    # TODO! we should poll for the registry socket-bind to take place
+    # and only once that's done yield to the requester!
+    # -[ ] TCP: use the `._root.open_root_actor()`::`ping_tpt_socket()`
+    #      closure!
+    # -[ ] UDS: can we do something similar for 'pinging" the
+    #     file-socket?
+    #
+    global _PROC_SPAWN_WAIT
     # UDS sockets are **really** fast to bind()/listen()/connect()
     # so it's often required that we delay a bit more starting
     # the first actor-tree..
     if tpt_proto == 'uds':
-        global _PROC_SPAWN_WAIT
-        _PROC_SPAWN_WAIT = 0.6
+        _PROC_SPAWN_WAIT += 1.6
 
+    if _non_linux and ci_env:
+        _PROC_SPAWN_WAIT += 1
+
+    # XXX, allow time for the sub-py-proc to boot up.
+    # !TODO, see ping-polling ideas above!
     time.sleep(_PROC_SPAWN_WAIT)
 
     assert not proc.returncode
@@ -163,18 +217,30 @@ def daemon(
     # XXX! yeah.. just be reaaal careful with this bc sometimes it
     # can lock up on the `_io.BufferedReader` and hang..
     stderr: str = proc.stderr.read().decode()
-    if stderr:
+    stdout: str = proc.stdout.read().decode()
+    if (
+        stderr
+        or
+        stdout
+    ):
         print(
-            f'Daemon actor tree produced STDERR:\n'
+            f'Daemon actor tree produced output:\n'
             f'{proc.args}\n'
             f'\n'
-            f'{stderr}\n'
+            f'stderr: {stderr!r}\n'
+            f'stdout: {stdout!r}\n'
         )
-    if proc.returncode != -2:
-        raise RuntimeError(
-            'Daemon actor tree failed !?\n'
-            f'{proc.args}\n'
+
+    if (rc := proc.returncode) != -2:
+        msg: str = (
+            f'Daemon actor tree was not cancelled !?\n'
+            f'proc.args: {proc.args!r}\n'
+            f'proc.returncode: {rc!r}\n'
         )
+        if rc < 0:
+            raise RuntimeError(msg)
+
+        test_log.error(msg)
 
 
 # @pytest.fixture(autouse=True)

tests/devx/conftest.py

@@ -3,8 +3,9 @@
 
 '''
 from __future__ import annotations
-import time
+import platform
 import signal
+import time
 from typing import (
     Callable,
     TYPE_CHECKING,
@@ -33,6 +34,17 @@ if TYPE_CHECKING:
     from pexpect import pty_spawn
 
 
+_non_linux: bool = platform.system() != 'Linux'
+
+
+def pytest_configure(config):
+    # register custom marks to avoid warnings see,
+    # https://docs.pytest.org/en/stable/how-to/writing_plugins.html#registering-custom-markers
+    config.addinivalue_line(
+        'markers',
+        'ctlcs_bish: test will (likely) not behave under SIGINT..'
+    )
+
 # a fn that sub-instantiates a `pexpect.spawn()`
 # and returns it.
 type PexpectSpawner = Callable[
@@ -68,12 +80,16 @@ def spawn(
 
         '''
         import os
+        # disable colored tbs
         os.environ['PYTHON_COLORS'] = '0'
+        # disable all ANSI color output
+        # os.environ['NO_COLOR'] = '1'
 
     spawned: PexpectSpawner|None = None
 
     def _spawn(
         cmd: str,
+        expect_timeout: float = 4,
         **mkcmd_kwargs,
     ) -> pty_spawn.spawn:
         nonlocal spawned
@@ -83,11 +99,17 @@ def spawn(
                 cmd,
                 **mkcmd_kwargs,
             ),
-            expect_timeout=3,
+            expect_timeout=(timeout:=(
+                expect_timeout + 6
+                if _non_linux and _ci_env
+                else expect_timeout
+            )),
             # preexec_fn=unset_colors,
             # ^TODO? get `pytest` core to expose underlying
             # `pexpect.spawn()` stuff?
         )
+        # sanity
+        assert spawned.timeout == timeout
         return spawned
 
     # such that test-dep can pass input script name.
@@ -146,6 +168,8 @@ def ctlc(
             mark.name == 'ctlcs_bish'
             and
             use_ctlc
+            and
+            all(mark.args)
         ):
             pytest.skip(
                 f'Test {node} prolly uses something from the stdlib (namely `asyncio`..)\n'
@@ -251,12 +275,13 @@ def assert_before(
         err_on_false=True,
         **kwargs
     )
+    return str(child.before.decode())
 
 
 def do_ctlc(
     child,
     count: int = 3,
-    delay: float = 0.1,
+    delay: float|None = None,
     patt: str|None = None,
 
     # expect repl UX to reprint the prompt after every
@@ -268,6 +293,7 @@ def do_ctlc(
 ) -> str|None:
 
     before: str|None = None
+    delay = delay or 0.1
 
     # make sure ctl-c sends don't do anything but repeat output
     for _ in range(count):
@@ -278,7 +304,10 @@ def do_ctlc(
         # if you run this test manually it works just fine..
         if expect_prompt:
             time.sleep(delay)
-            child.expect(PROMPT)
+            child.expect(
+                PROMPT,
+                timeout=(child.timeout * 2) if _ci_env else child.timeout,
+            )
             before = str(child.before.decode())
             time.sleep(delay)
 

tests/devx/test_debugger.py

@@ -37,6 +37,9 @@ from .conftest import (
     in_prompt_msg,
     assert_before,
 )
+from ..conftest import (
+    _ci_env,
+)
 
 if TYPE_CHECKING:
     from ..conftest import PexpectSpawner
@@ -51,13 +54,14 @@ if TYPE_CHECKING:
 # - recurrent root errors
 
 
+_non_linux: bool = platform.system() != 'Linux'
+
 if platform.system() == 'Windows':
     pytest.skip(
         'Debugger tests have no windows support (yet)',
         allow_module_level=True,
     )
 
-
 # TODO: was trying to this xfail style but some weird bug i see in CI
 # that's happening at collect time.. pretty soon gonna dump actions i'm
 # thinkin...
@@ -193,6 +197,11 @@ def test_root_actor_bp_forever(
     child.expect(EOF)
 
 
+# skip on non-Linux CI
+@pytest.mark.ctlcs_bish(
+    _non_linux,
+    _ci_env,
+)
 @pytest.mark.parametrize(
     'do_next',
     (True, False),
@@ -258,6 +267,11 @@ def test_subactor_error(
     child.expect(EOF)
 
 
+# skip on non-Linux CI
+@pytest.mark.ctlcs_bish(
+    _non_linux,
+    _ci_env,
+)
 def test_subactor_breakpoint(
     spawn,
     ctlc: bool,
@@ -480,8 +494,24 @@ def test_multi_daemon_subactors(
     stream.
 
     '''
-    child = spawn('multi_daemon_subactors')
+    non_linux = _non_linux
+    if non_linux and ctlc:
+        pytest.skip(
+            'Ctl-c + MacOS is too unreliable/racy for this test..\n'
+        )
+        # !TODO, if someone with more patience then i wants to muck
+        # with the timings on this please feel free to see all the
+        # `non_linux` branching logic i added on my first attempt
+        # below!
+        #
+        # my conclusion was that if i were to run the script
+        # manually, and thus as slowly as a human would, the test
+        # would and should pass as described in this test fn, however
+        # after fighting with it for >= 1hr. i decided more then
+        # likely the more extensive `linux` testing should cover most
+        # regressions.
 
+    child = spawn('multi_daemon_subactors')
     child.expect(PROMPT)
 
     # there can be a race for which subactor will acquire
@@ -511,8 +541,19 @@ def test_multi_daemon_subactors(
     else:
         raise ValueError('Neither log msg was found !?')
 
+    non_linux_delay: float = 0.3
     if ctlc:
-        do_ctlc(child)
+        do_ctlc(
+            child,
+            delay=(
+                non_linux_delay
+                if non_linux
+                else None
+            ),
+        )
+
+        if non_linux:
+            time.sleep(1)
 
     # NOTE: previously since we did not have clobber prevention
     # in the root actor this final resume could result in the debugger
@@ -543,33 +584,66 @@ def test_multi_daemon_subactors(
     # assert "in use by child ('bp_forever'," in before
 
     if ctlc:
-        do_ctlc(child)
+        do_ctlc(
+            child,
+            delay=(
+                non_linux_delay
+                if non_linux
+                else None
+            ),
+        )
+
+        if non_linux:
+            time.sleep(1)
 
     # expect another breakpoint actor entry
     child.sendline('c')
     child.expect(PROMPT)
-
     try:
-        assert_before(
+        before: str = assert_before(
             child,
             bp_forev_parts,
         )
     except AssertionError:
-        assert_before(
+        before: str = assert_before(
             child,
             name_error_parts,
         )
 
     else:
         if ctlc:
-            do_ctlc(child)
+            before: str = do_ctlc(
+                child,
+                delay=(
+                    non_linux_delay
+                    if non_linux
+                    else None
+                ),
+            )
+
+            if non_linux:
+                time.sleep(1)
 
         # should crash with the 2nd name error (simulates
         # a retry) and then the root eventually (boxed) errors
         # after 1 or more further bp actor entries.
 
         child.sendline('c')
-        child.expect(PROMPT)
+        try:
+            child.expect(
+                PROMPT,
+                timeout=3,
+            )
+        except EOF:
+            before: str = child.before.decode()
+            print(
+                f'\n'
+                f'??? NEVER RXED `pdb` PROMPT ???\n'
+                f'\n'
+                f'{before}\n'
+            )
+            raise
+
         assert_before(
             child,
             name_error_parts,
@@ -689,7 +763,8 @@ def test_multi_subactors_root_errors(
 
 @has_nested_actors
 def test_multi_nested_subactors_error_through_nurseries(
-    spawn,
+    ci_env: bool,
+    spawn: PexpectSpawner,
 
     # TODO: address debugger issue for nested tree:
     # https://github.com/goodboy/tractor/issues/320
@@ -712,7 +787,16 @@ def test_multi_nested_subactors_error_through_nurseries(
 
     for send_char in itertools.cycle(['c', 'q']):
         try:
-            child.expect(PROMPT)
+            child.expect(
+                PROMPT,
+                timeout=(
+                    6 if (
+                        _non_linux
+                        and
+                        ci_env
+                    ) else -1
+                ),
+            )
             child.sendline(send_char)
             time.sleep(0.01)
 
@@ -889,6 +973,11 @@ def test_different_debug_mode_per_actor(
     )
 
 
+# skip on non-Linux CI
+@pytest.mark.ctlcs_bish(
+    _non_linux,
+    _ci_env,
+)
 def test_post_mortem_api(
     spawn,
     ctlc: bool,
@@ -1133,14 +1222,20 @@ def test_ctxep_pauses_n_maybe_ipc_breaks(
             # closed so verify we see error reporting as well as
             # a failed crash-REPL request msg and can CTL-c our way
             # out.
+
+            # ?TODO, match depending on `tpt_proto(s)`?
+            # - [ ] how can we pass it into the script tho?
+            tpt: str = 'UDS'
+            if _non_linux:
+                tpt: str = 'TCP'
+
             assert_before(
                 child,
                 ['peer IPC channel closed abruptly?',
                  'another task closed this fd',
                  'Debug lock request was CANCELLED?',
-                 "'MsgpackUDSStream' was already closed locally?",
-                 "TransportClosed: 'MsgpackUDSStream' was already closed 'by peer'?",
-                 # ?TODO^? match depending on `tpt_proto(s)`?
+                 f"'Msgpack{tpt}Stream' was already closed locally?",
+                 f"TransportClosed: 'Msgpack{tpt}Stream' was already closed 'by peer'?",
                 ]
 
                 # XXX races on whether these show/hit?

tests/devx/test_tooling.py

@@ -31,6 +31,9 @@ from .conftest import (
     PROMPT,
     _pause_msg,
 )
+from ..conftest import (
+    no_macos,
+)
 
 import pytest
 from pexpect.exceptions import (
@@ -42,6 +45,7 @@ if TYPE_CHECKING:
     from ..conftest import PexpectSpawner
 
 
+@no_macos
 def test_shield_pause(
     spawn: PexpectSpawner,
 ):
@@ -57,6 +61,7 @@ def test_shield_pause(
     expect(
         child,
         'Yo my child hanging..?',
+        timeout=3,
     )
     assert_before(
         child,

tests/ipc/test_multi_tpt.py

@@ -62,6 +62,13 @@ def test_root_passes_tpt_to_sub(
     reg_addr: tuple,
     debug_mode: bool,
 ):
+    # XXX NOTE, the `reg_addr` addr won't be the same type as the
+    # `tpt_proto_key` would deliver here unless you pass `--tpt-proto
+    # <tpt_proto_key>` on the CLI.
+    #
+    # if tpt_proto_key == 'uds':
+    #     breakpoint()
+
     async def main():
         async with tractor.open_nursery(
             enable_transports=[tpt_proto_key],

tests/msg/__init__.py

@@ -0,0 +1,4 @@
+'''
+`tractor.msg.*` sub-sys test suite.
+
+'''

tests/msg/conftest.py

@@ -0,0 +1,4 @@
+'''
+`tractor.msg.*` test sub-pkg conf.
+
+'''

tests/msg/test_pretty_struct.py

@@ -0,0 +1,240 @@
+'''
+Unit tests for `tractor.msg.pretty_struct`
+private-field filtering in `pformat()`.
+
+'''
+import pytest
+
+from tractor.msg.pretty_struct import (
+    Struct,
+    pformat,
+    iter_struct_ppfmt_lines,
+)
+from tractor.msg._codec import (
+    MsgDec,
+    mk_dec,
+)
+
+
+# ------ test struct definitions ------ #
+
+class PublicOnly(Struct):
+    '''
+    All-public fields for baseline testing.
+
+    '''
+    name: str = 'alice'
+    age: int = 30
+
+
+class PrivateOnly(Struct):
+    '''
+    Only underscore-prefixed (private) fields.
+
+    '''
+    _secret: str = 'hidden'
+    _internal: int = 99
+
+
+class MixedFields(Struct):
+    '''
+    Mix of public and private fields.
+
+    '''
+    name: str = 'bob'
+    _hidden: int = 42
+    value: float = 3.14
+    _meta: str = 'internal'
+
+
+class Inner(
+    Struct,
+    frozen=True,
+):
+    '''
+    Frozen inner struct with a private field,
+    for nesting tests.
+
+    '''
+    x: int = 1
+    _secret: str = 'nope'
+
+
+class Outer(Struct):
+    '''
+    Outer struct nesting an `Inner`.
+
+    '''
+    label: str = 'outer'
+    inner: Inner = Inner()
+
+
+class EmptyStruct(Struct):
+    '''
+    Struct with zero fields.
+
+    '''
+    pass
+
+
+# ------ tests ------ #
+
+@pytest.mark.parametrize(
+    'struct_and_expected',
+    [
+        (
+            PublicOnly(),
+            {
+                'shown': ['name', 'age'],
+                'hidden': [],
+            },
+        ),
+        (
+            MixedFields(),
+            {
+                'shown': ['name', 'value'],
+                'hidden': ['_hidden', '_meta'],
+            },
+        ),
+        (
+            PrivateOnly(),
+            {
+                'shown': [],
+                'hidden': ['_secret', '_internal'],
+            },
+        ),
+    ],
+    ids=[
+        'all-public',
+        'mixed-pub-priv',
+        'all-private',
+    ],
+)
+def test_field_visibility_in_pformat(
+    struct_and_expected: tuple[
+        Struct,
+        dict[str, list[str]],
+    ],
+):
+    '''
+    Verify `pformat()` shows public fields
+    and hides `_`-prefixed private fields.
+
+    '''
+    (
+        struct,
+        expected,
+    ) = struct_and_expected
+    output: str = pformat(struct)
+
+    for field_name in expected['shown']:
+        assert field_name in output, (
+            f'{field_name!r} should appear in:\n'
+            f'{output}'
+        )
+
+    for field_name in expected['hidden']:
+        assert field_name not in output, (
+            f'{field_name!r} should NOT appear in:\n'
+            f'{output}'
+        )
+
+
+def test_iter_ppfmt_lines_skips_private():
+    '''
+    Directly verify `iter_struct_ppfmt_lines()`
+    never yields tuples with `_`-prefixed field
+    names.
+
+    '''
+    struct = MixedFields()
+    lines: list[tuple[str, str]] = list(
+        iter_struct_ppfmt_lines(
+            struct,
+            field_indent=2,
+        )
+    )
+    # should have lines for public fields only
+    assert len(lines) == 2
+
+    for _prefix, line_content in lines:
+        field_name: str = (
+            line_content.split(':')[0].strip()
+        )
+        assert not field_name.startswith('_'), (
+            f'private field leaked: {field_name!r}'
+        )
+
+
+def test_nested_struct_filters_inner_private():
+    '''
+    Verify that nested struct's private fields
+    are also filtered out during recursion.
+
+    '''
+    outer = Outer()
+    output: str = pformat(outer)
+
+    # outer's public field
+    assert 'label' in output
+
+    # inner's public field (recursed into)
+    assert 'x' in output
+
+    # inner's private field must be hidden
+    assert '_secret' not in output
+
+
+def test_empty_struct_pformat():
+    '''
+    An empty struct should produce a valid
+    `pformat()` result with no field lines.
+
+    '''
+    output: str = pformat(EmptyStruct())
+    assert 'EmptyStruct(' in output
+    assert output.rstrip().endswith(')')
+
+    # no field lines => only struct header+footer
+    lines: list[tuple[str, str]] = list(
+        iter_struct_ppfmt_lines(
+            EmptyStruct(),
+            field_indent=2,
+        )
+    )
+    assert lines == []
+
+
+def test_real_msgdec_pformat_hides_private():
+    '''
+    Verify `pformat()` on a real `MsgDec`
+    hides the `_dec` internal field.
+
+    NOTE: `MsgDec.__repr__` is custom and does
+    NOT call `pformat()`, so we call it directly.
+
+    '''
+    dec: MsgDec = mk_dec(spec=int)
+    output: str = pformat(dec)
+
+    # the private `_dec` field should be filtered
+    assert '_dec' not in output
+
+    # but the struct type name should be present
+    assert 'MsgDec(' in output
+
+
+def test_pformat_repr_integration():
+    '''
+    Verify that `Struct.__repr__()` (which calls
+    `pformat()`) also hides private fields for
+    custom structs that do NOT override `__repr__`.
+
+    '''
+    mixed = MixedFields()
+    output: str = repr(mixed)
+
+    assert 'name' in output
+    assert 'value' in output
+    assert '_hidden' not in output
+    assert '_meta' not in output

tests/test_2way.py

@@ -1,7 +1,12 @@
-"""
-Bidirectional streaming.
+'''
+Audit the simplest inter-actor bidirectional (streaming)
+msg patterns.
 
-"""
+'''
+from __future__ import annotations
+from typing import (
+    Callable,
+)
 import pytest
 import trio
 import tractor
@@ -9,10 +14,8 @@ import tractor
 
 @tractor.context
 async def simple_rpc(
-
     ctx: tractor.Context,
     data: int,
-
 ) -> None:
     '''
     Test a small ping-pong server.
@@ -39,15 +42,13 @@ async def simple_rpc(
 
 @tractor.context
 async def simple_rpc_with_forloop(
-
     ctx: tractor.Context,
     data: int,
-
 ) -> None:
-    """Same as previous test but using ``async for`` syntax/api.
-
-    """
+    '''
+    Same as previous test but using `async for` syntax/api.
 
+    '''
     # signal to parent that we're up
     await ctx.started(data + 1)
 
@@ -68,62 +69,78 @@ async def simple_rpc_with_forloop(
 
 @pytest.mark.parametrize(
     'use_async_for',
-    [True, False],
+    [
+        True,
+        False,
+    ],
+    ids='use_async_for={}'.format,
 )
 @pytest.mark.parametrize(
     'server_func',
-    [simple_rpc, simple_rpc_with_forloop],
+    [
+        simple_rpc,
+        simple_rpc_with_forloop,
+    ],
+    ids='server_func={}'.format,
 )
-def test_simple_rpc(server_func, use_async_for):
+def test_simple_rpc(
+    server_func: Callable,
+    use_async_for: bool,
+    loglevel: str,
+    debug_mode: bool,
+):
     '''
     The simplest request response pattern.
 
     '''
     async def main():
-        async with tractor.open_nursery() as n:
+        with trio.fail_after(6):
+            async with tractor.open_nursery(
+                loglevel=loglevel,
+                debug_mode=debug_mode,
+            ) as an:
+                portal: tractor.Portal = await an.start_actor(
+                    'rpc_server',
+                    enable_modules=[__name__],
+                )
 
-            portal = await n.start_actor(
-                'rpc_server',
-                enable_modules=[__name__],
-            )
+                async with portal.open_context(
+                    server_func,  # taken from pytest parameterization
+                    data=10,
+                ) as (ctx, sent):
 
-            async with portal.open_context(
-                server_func,  # taken from pytest parameterization
-                data=10,
-            ) as (ctx, sent):
+                    assert sent == 11
 
-                assert sent == 11
+                    async with ctx.open_stream() as stream:
 
-                async with ctx.open_stream() as stream:
+                        if use_async_for:
 
-                    if use_async_for:
-
-                        count = 0
-                        # receive msgs using async for style
-                        print('ping')
-                        await stream.send('ping')
-
-                        async for msg in stream:
-                            assert msg == 'pong'
+                            count = 0
+                            # receive msgs using async for style
                             print('ping')
                             await stream.send('ping')
-                            count += 1
 
-                            if count >= 9:
-                                break
+                            async for msg in stream:
+                                assert msg == 'pong'
+                                print('ping')
+                                await stream.send('ping')
+                                count += 1
 
-                    else:
-                        # classic send/receive style
-                        for _ in range(10):
+                                if count >= 9:
+                                    break
 
-                            print('ping')
-                            await stream.send('ping')
-                            assert await stream.receive() == 'pong'
+                        else:
+                            # classic send/receive style
+                            for _ in range(10):
 
-                # stream should terminate here
+                                print('ping')
+                                await stream.send('ping')
+                                assert await stream.receive() == 'pong'
 
-            # final context result(s) should be consumed here in __aexit__()
+                    # stream should terminate here
 
-            await portal.cancel_actor()
+                # final context result(s) should be consumed here in __aexit__()
+
+                await portal.cancel_actor()
 
     trio.run(main)

tests/test_advanced_faults.py

@@ -146,9 +146,6 @@ def test_ipc_channel_break_during_stream(
         # a user sending ctl-c by raising a KBI.
         if pre_aclose_msgstream:
             expect_final_exc = KeyboardInterrupt
-            if tpt_proto == 'uds':
-                expect_final_exc = TransportClosed
-                expect_final_cause = trio.BrokenResourceError
 
             # XXX OLD XXX
             # if child calls `MsgStream.aclose()` then expect EoC.
@@ -169,10 +166,6 @@ def test_ipc_channel_break_during_stream(
         if pre_aclose_msgstream:
             expect_final_exc = KeyboardInterrupt
 
-            if tpt_proto == 'uds':
-                expect_final_exc = TransportClosed
-                expect_final_cause = trio.BrokenResourceError
-
     # NOTE when the parent IPC side dies (even if the child does as well
     # but the child fails BEFORE the parent) we always expect the
     # IPC layer to raise a closed-resource, NEVER do we expect

tests/test_cancellation.py

@@ -17,8 +17,8 @@ from tractor._testing import (
 from .conftest import no_windows
 
 
-def is_win():
-    return platform.system() == 'Windows'
+_non_linux: bool = platform.system() != 'Linux'
+_friggin_windows: bool = platform.system() == 'Windows'
 
 
 async def assert_err(delay=0):
@@ -431,7 +431,7 @@ async def test_nested_multierrors(loglevel, start_method):
             for subexc in err.exceptions:
 
                 # verify first level actor errors are wrapped as remote
-                if is_win():
+                if _friggin_windows:
 
                     # windows is often too slow and cancellation seems
                     # to happen before an actor is spawned
@@ -464,7 +464,7 @@ async def test_nested_multierrors(loglevel, start_method):
                     # XXX not sure what's up with this..
                     # on windows sometimes spawning is just too slow and
                     # we get back the (sent) cancel signal instead
-                    if is_win():
+                    if _friggin_windows:
                         if isinstance(subexc, tractor.RemoteActorError):
                             assert subexc.boxed_type in (
                                 BaseExceptionGroup,
@@ -507,17 +507,22 @@ def test_cancel_via_SIGINT(
 
 @no_windows
 def test_cancel_via_SIGINT_other_task(
-    loglevel,
-    start_method,
-    spawn_backend,
+    loglevel: str,
+    start_method: str,
+    spawn_backend: str,
 ):
-    """Ensure that a control-C (SIGINT) signal cancels both the parent
-    and child processes in trionic fashion even a subprocess is started
-    from a seperate ``trio`` child  task.
-    """
-    pid = os.getpid()
-    timeout: float = 2
-    if is_win():  # smh
+    '''
+    Ensure that a control-C (SIGINT) signal cancels both the parent
+    and child processes in trionic fashion even a subprocess is
+    started from a seperate ``trio`` child  task.
+
+    '''
+    pid: int = os.getpid()
+    timeout: float = (
+        4 if _non_linux
+        else 2
+    )
+    if _friggin_windows:  # smh
         timeout += 1
 
     async def spawn_and_sleep_forever(
@@ -644,7 +649,11 @@ def test_cancel_while_childs_child_in_sync_sleep(
         #
         # delay = 1  # no AssertionError in eg, TooSlowError raised.
         # delay = 2  # is AssertionError in eg AND no TooSlowError !?
-        delay = 4  # is AssertionError in eg AND no _cs cancellation.
+        # is AssertionError in eg AND no _cs cancellation.
+        delay = (
+            6 if _non_linux
+            else 4 
+        )
 
         with trio.fail_after(delay) as _cs:
         # with trio.CancelScope() as cs:
@@ -696,7 +705,7 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon(
     kbi_delay = 0.5
     timeout: float = 2.9
 
-    if is_win():  # smh
+    if _friggin_windows:  # smh
         timeout += 1
 
     async def main():

tests/test_child_manages_service_nursery.py

@@ -18,16 +18,15 @@ from tractor import RemoteActorError
 
 
 async def aio_streamer(
-    from_trio: asyncio.Queue,
-    to_trio: trio.abc.SendChannel,
+    chan: tractor.to_asyncio.LinkedTaskChannel,
 ) -> trio.abc.ReceiveChannel:
 
     # required first msg to sync caller
-    to_trio.send_nowait(None)
+    chan.started_nowait(None)
 
     from itertools import cycle
     for i in cycle(range(10)):
-        to_trio.send_nowait(i)
+        chan.send_nowait(i)
         await asyncio.sleep(0.01)
 
 
@@ -69,7 +68,7 @@ async def wrapper_mngr(
         else:
             async with tractor.to_asyncio.open_channel_from(
                 aio_streamer,
-            ) as (first, from_aio):
+            ) as (from_aio, first):
                 assert not first
 
                 # cache it so next task uses broadcast receiver

tests/test_clustering.py

@@ -11,7 +11,6 @@ MESSAGE = 'tractoring at full speed'
 
 
 def test_empty_mngrs_input_raises() -> None:
-
     async def main():
         with trio.fail_after(3):
             async with (
@@ -56,25 +55,39 @@ async def worker(
             print(msg)
             assert msg == MESSAGE
 
-        # TODO: does this ever cause a hang
+        # ?TODO, does this ever cause a hang?
         # assert 0
 
 
+# ?TODO, but needs a fn-scoped tpt_proto fixture..
+# @pytest.mark.no_tpt('uds')
 @tractor_test
-async def test_streaming_to_actor_cluster() -> None:
+async def test_streaming_to_actor_cluster(
+    tpt_proto: str,
+):
+    '''
+    Open an actor "cluster" using the (experimental) `._clustering`
+    API and conduct standard inter-task-ctx streaming.
 
-    async with (
-        open_actor_cluster(modules=[__name__]) as portals,
+    '''
+    if tpt_proto == 'uds':
+        pytest.skip(
+            f'Test currently fails with tpt-proto={tpt_proto!r}\n'
+        )
 
-        gather_contexts(
-            mngrs=[p.open_context(worker) for p in portals.values()],
-        ) as contexts,
+    with trio.fail_after(6):
+        async with (
+            open_actor_cluster(modules=[__name__]) as portals,
 
-        gather_contexts(
-            mngrs=[ctx[0].open_stream() for ctx in contexts],
-        ) as streams,
+            gather_contexts(
+                mngrs=[p.open_context(worker) for p in portals.values()],
+            ) as contexts,
 
-    ):
-        with trio.move_on_after(1):
-            for stream in itertools.cycle(streams):
-                await stream.send(MESSAGE)
+            gather_contexts(
+                mngrs=[ctx[0].open_stream() for ctx in contexts],
+            ) as streams,
+
+        ):
+            with trio.move_on_after(1):
+                for stream in itertools.cycle(streams):
+                    await stream.send(MESSAGE)

tests/test_context_stream_semantics.py

@@ -9,6 +9,7 @@ from itertools import count
 import math
 import platform
 from pprint import pformat
+import sys
 from typing import (
     Callable,
 )
@@ -941,6 +942,11 @@ def test_one_end_stream_not_opened(
     from tractor._runtime import Actor
     buf_size = buf_size_increase + Actor.msg_buffer_size
 
+    timeout: float = (
+        1 if sys.platform == 'linux'
+        else 3
+    )
+
     async def main():
         async with tractor.open_nursery(
             debug_mode=debug_mode,
@@ -950,7 +956,7 @@ def test_one_end_stream_not_opened(
                 enable_modules=[__name__],
             )
 
-            with trio.fail_after(1):
+            with trio.fail_after(timeout):
                 async with portal.open_context(
                     entrypoint,
                 ) as (ctx, sent):

tests/test_discovery.py

@@ -1,11 +1,14 @@
 """
-Actor "discovery" testing
+Discovery subsys.
+
 """
 import os
 import signal
 import platform
 from functools import partial
 import itertools
+import time
+from typing import Callable
 
 import psutil
 import pytest
@@ -17,7 +20,9 @@ import trio
 
 
 @tractor_test
-async def test_reg_then_unreg(reg_addr):
+async def test_reg_then_unreg(
+    reg_addr: tuple,
+):
     actor = tractor.current_actor()
     assert actor.is_arbiter
     assert len(actor._registry) == 1  # only self is registered
@@ -27,7 +32,7 @@ async def test_reg_then_unreg(reg_addr):
     ) as n:
 
         portal = await n.start_actor('actor', enable_modules=[__name__])
-        uid = portal.channel.uid
+        uid = portal.channel.aid.uid
 
         async with tractor.get_registry(reg_addr) as aportal:
             # this local actor should be the arbiter
@@ -82,11 +87,15 @@ async def say_hello_use_wait(
 
 
 @tractor_test
-@pytest.mark.parametrize('func', [say_hello, say_hello_use_wait])
+@pytest.mark.parametrize(
+    'func',
+    [say_hello,
+     say_hello_use_wait]
+)
 async def test_trynamic_trio(
-    func,
-    start_method,
-    reg_addr,
+    func: Callable,
+    start_method: str,
+    reg_addr: tuple,
 ):
     '''
     Root actor acting as the "director" and running one-shot-task-actors
@@ -119,7 +128,10 @@ async def stream_forever():
         await trio.sleep(0.01)
 
 
-async def cancel(use_signal, delay=0):
+async def cancel(
+    use_signal: bool,
+    delay: float = 0,
+):
     # hold on there sally
     await trio.sleep(delay)
 
@@ -132,13 +144,15 @@ async def cancel(use_signal, delay=0):
         raise KeyboardInterrupt
 
 
-async def stream_from(portal):
+async def stream_from(portal: tractor.Portal):
     async with portal.open_stream_from(stream_forever) as stream:
         async for value in stream:
             print(value)
 
 
-async def unpack_reg(actor_or_portal):
+async def unpack_reg(
+    actor_or_portal: tractor.Portal|tractor.Actor,
+):
     '''
     Get and unpack a "registry" RPC request from the "arbiter" registry
     system.
@@ -173,7 +187,9 @@ async def spawn_and_check_registry(
         registry_addrs=[reg_addr],
         debug_mode=debug_mode,
     ):
-        async with tractor.get_registry(reg_addr) as portal:
+        async with tractor.get_registry(
+            addr=reg_addr,
+        ) as portal:
             # runtime needs to be up to call this
             actor = tractor.current_actor()
 
@@ -190,7 +206,7 @@ async def spawn_and_check_registry(
 
             # ensure current actor is registered
             registry: dict = await get_reg()
-            assert actor.uid in registry
+            assert actor.aid.uid in registry
 
             try:
                 async with tractor.open_nursery() as an:
@@ -238,18 +254,31 @@ async def spawn_and_check_registry(
 
                 # all subactors should have de-registered
                 registry = await get_reg()
-                assert len(registry) == extra
-                assert actor.uid in registry
+                start: float = time.time()
+                while (
+                    not (len(registry) == extra)
+                    and
+                    (time.time() - start) < 5
+                ):
+                    print(
+                        f'Waiting for remaining subs to dereg..\n'
+                        f'{registry!r}\n'
+                    )
+                    await trio.sleep(0.3)
+                else:
+                    assert len(registry) == extra
+
+                assert actor.aid.uid in registry
 
 
 @pytest.mark.parametrize('use_signal', [False, True])
 @pytest.mark.parametrize('with_streaming', [False, True])
 def test_subactors_unregister_on_cancel(
     debug_mode: bool,
-    start_method,
-    use_signal,
-    reg_addr,
-    with_streaming,
+    start_method: str,
+    use_signal: bool,
+    reg_addr: tuple,
+    with_streaming: bool,
 ):
     '''
     Verify that cancelling a nursery results in all subactors
@@ -274,15 +303,17 @@ def test_subactors_unregister_on_cancel(
 def test_subactors_unregister_on_cancel_remote_daemon(
     daemon: subprocess.Popen,
     debug_mode: bool,
-    start_method,
-    use_signal,
-    reg_addr,
-    with_streaming,
+    start_method: str,
+    use_signal: bool,
+    reg_addr: tuple,
+    with_streaming: bool,
 ):
-    """Verify that cancelling a nursery results in all subactors
-    deregistering themselves with a **remote** (not in the local process
-    tree) arbiter.
-    """
+    '''
+    Verify that cancelling a nursery results in all subactors
+    deregistering themselves with a **remote** (not in the local
+    process tree) arbiter.
+
+    '''
     with pytest.raises(KeyboardInterrupt):
         trio.run(
             partial(
@@ -367,21 +398,23 @@ async def close_chans_before_nursery(
 
                     # all subactors should have de-registered
                     registry = await get_reg()
-                    assert portal1.channel.uid not in registry
-                    assert portal2.channel.uid not in registry
+                    assert portal1.channel.aid.uid not in registry
+                    assert portal2.channel.aid.uid not in registry
                     assert len(registry) == entries_at_end
 
 
 @pytest.mark.parametrize('use_signal', [False, True])
 def test_close_channel_explicit(
-    start_method,
-    use_signal,
-    reg_addr,
+    start_method: str,
+    use_signal: bool,
+    reg_addr: tuple,
 ):
-    """Verify that closing a stream explicitly and killing the actor's
+    '''
+    Verify that closing a stream explicitly and killing the actor's
     "root nursery" **before** the containing nursery tears down also
     results in subactor(s) deregistering from the arbiter.
-    """
+
+    '''
     with pytest.raises(KeyboardInterrupt):
         trio.run(
             partial(
@@ -396,14 +429,16 @@ def test_close_channel_explicit(
 @pytest.mark.parametrize('use_signal', [False, True])
 def test_close_channel_explicit_remote_arbiter(
     daemon: subprocess.Popen,
-    start_method,
-    use_signal,
-    reg_addr,
+    start_method: str,
+    use_signal: bool,
+    reg_addr: tuple,
 ):
-    """Verify that closing a stream explicitly and killing the actor's
+    '''
+    Verify that closing a stream explicitly and killing the actor's
     "root nursery" **before** the containing nursery tears down also
     results in subactor(s) deregistering from the arbiter.
-    """
+
+    '''
     with pytest.raises(KeyboardInterrupt):
         trio.run(
             partial(

tests/test_docs_examples.py

@@ -9,12 +9,17 @@ import sys
 import subprocess
 import platform
 import shutil
+from typing import Callable
 
 import pytest
+import tractor
 from tractor._testing import (
     examples_dir,
 )
 
+_non_linux: bool = platform.system() != 'Linux'
+_friggin_macos: bool = platform.system() == 'Darwin'
+
 
 @pytest.fixture
 def run_example_in_subproc(
@@ -101,8 +106,10 @@ def run_example_in_subproc(
     ids=lambda t: t[1],
 )
 def test_example(
-    run_example_in_subproc,
-    example_script,
+    run_example_in_subproc: Callable,
+    example_script: str,
+    test_log: tractor.log.StackLevelAdapter,
+    ci_env: bool,
 ):
     '''
     Load and run scripts from this repo's ``examples/`` dir as a user
@@ -116,9 +123,32 @@ def test_example(
     '''
     ex_file: str = os.path.join(*example_script)
 
-    if 'rpc_bidir_streaming' in ex_file and sys.version_info < (3, 9):
+    if (
+        'rpc_bidir_streaming' in ex_file
+        and
+        sys.version_info < (3, 9)
+    ):
         pytest.skip("2-way streaming example requires py3.9 async with syntax")
 
+    if (
+        'full_fledged_streaming_service' in ex_file
+        and
+        _friggin_macos
+        and
+        ci_env
+    ):
+        pytest.skip(
+            'Streaming example is too flaky in CI\n'
+            'AND their competitor runs this CI service..\n'
+            'This test does run just fine "in person" however..'
+        )
+
+    timeout: float = (
+        60
+        if ci_env and _non_linux
+        else 16
+    )
+
     with open(ex_file, 'r') as ex:
         code = ex.read()
 
@@ -126,9 +156,12 @@ def test_example(
             err = None
             try:
                 if not proc.poll():
-                    _, err = proc.communicate(timeout=15)
+                    _, err = proc.communicate(timeout=timeout)
 
             except subprocess.TimeoutExpired as e:
+                test_log.exception(
+                    f'Example failed to finish within {timeout}s ??\n'
+                )
                 proc.kill()
                 err = e.stderr
 

tests/test_infected_asyncio.py

@@ -47,12 +47,11 @@ async def sleep_and_err(
 
     # just signature placeholders for compat with
     # ``to_asyncio.open_channel_from()``
-    to_trio: trio.MemorySendChannel|None = None,
-    from_trio: asyncio.Queue|None = None,
+    chan: to_asyncio.LinkedTaskChannel|None = None,
 
 ):
-    if to_trio:
-        to_trio.send_nowait('start')
+    if chan:
+        chan.started_nowait('start')
 
     await asyncio.sleep(sleep_for)
     assert 0
@@ -238,7 +237,7 @@ async def trio_ctx(
                 trio.open_nursery() as tn,
                 tractor.to_asyncio.open_channel_from(
                     sleep_and_err,
-                ) as (first, chan),
+                ) as (chan, first),
             ):
 
                 assert first == 'start'
@@ -399,7 +398,7 @@ async def no_to_trio_in_args():
 
 async def push_from_aio_task(
     sequence: Iterable,
-    to_trio: trio.abc.SendChannel,
+    chan: to_asyncio.LinkedTaskChannel,
     expect_cancel: False,
     fail_early: bool,
     exit_early: bool,
@@ -407,15 +406,12 @@ async def push_from_aio_task(
 ) -> None:
 
     try:
-        # print('trying breakpoint')
-        # breakpoint()
-
         # sync caller ctx manager
-        to_trio.send_nowait(True)
+        chan.started_nowait(True)
 
         for i in sequence:
             print(f'asyncio sending {i}')
-            to_trio.send_nowait(i)
+            chan.send_nowait(i)
             await asyncio.sleep(0.001)
 
             if (
@@ -478,7 +474,7 @@ async def stream_from_aio(
                 trio_exit_early
             ))
 
-        ) as (first, chan):
+        ) as (chan, first):
 
             assert first is True
 
@@ -772,8 +768,8 @@ async def trio_to_aio_echo_server(
     async with to_asyncio.open_channel_from(
         aio_echo_server,
     ) as (
-        first,  # value from `chan.started_nowait()` above
         chan,
+        first,  # value from `chan.started_nowait()` above
     ):
         assert first == 'start'
 
@@ -1109,13 +1105,12 @@ async def raise_before_started(
 ) -> None:
     '''
     `asyncio.Task` entry point which RTEs before calling
-    `to_trio.send_nowait()`.
+    `chan.started_nowait()`.
 
     '''
     await asyncio.sleep(0.2)
     raise RuntimeError('Some shite went wrong before `.send_nowait()`!!')
 
-    # to_trio.send_nowait('Uhh we shouldve RTE-d ^^ ??')
     chan.started_nowait('Uhh we shouldve RTE-d ^^ ??')
     await asyncio.sleep(float('inf'))
 

tests/test_inter_peer_cancellation.py

@@ -581,7 +581,7 @@ def test_peer_canceller(
                             assert (
                                 re.canceller
                                 ==
-                                root.uid
+                                root.aid.uid
                             )
 
                         else:  # the other 2 ctxs
@@ -590,7 +590,7 @@ def test_peer_canceller(
                                 and (
                                     re.canceller
                                     ==
-                                    canceller.channel.uid
+                                    canceller.channel.aid.uid
                                 )
                             )
 
@@ -745,7 +745,7 @@ def test_peer_canceller(
                         # -> each context should have received
                         # a silently absorbed context cancellation
                         # in its remote nursery scope.
-                        # assert ctx.chan.uid == ctx.canceller
+                        # assert ctx.chan.aid.uid == ctx.canceller
 
                     # NOTE: when an inter-peer cancellation
                     # occurred, we DO NOT expect this
@@ -802,7 +802,7 @@ async def basic_echo_server(
 
     '''
     actor: Actor = tractor.current_actor()
-    uid: tuple = actor.uid
+    uid: tuple = actor.aid.uid
     await ctx.started(uid)
     async with ctx.open_stream() as ipc:
         async for msg in ipc:
@@ -857,7 +857,7 @@ async def serve_subactors(
                     f'|_{peer}\n'
                 )
                 await ipc.send((
-                    peer.chan.uid,
+                    peer.chan.aid.uid,
                     peer.chan.raddr.unwrap(),
                 ))
 
@@ -992,7 +992,7 @@ async def tell_little_bro(
         sub_ctx.open_stream() as echo_ipc,
     ):
         actor: Actor = current_actor()
-        uid: tuple = actor.uid
+        uid: tuple = actor.aid.uid
         for i in range(rng_seed):
             msg: tuple = (
                 uid,
@@ -1097,7 +1097,7 @@ def test_peer_spawns_and_cancels_service_subactor(
                     ) as (client_ctx, client_says),
                 ):
                     root: Actor = current_actor()
-                    spawner_uid: tuple = spawn_ctx.chan.uid
+                    spawner_uid: tuple = spawn_ctx.chan.aid.uid
                     print(
                         f'Server says: {first}\n'
                         f'Client says: {client_says}\n'
@@ -1116,7 +1116,7 @@ def test_peer_spawns_and_cancels_service_subactor(
                     print(
                         'Sub-spawn came online\n'
                         f'portal: {sub}\n'
-                        f'.uid: {sub.actor.uid}\n'
+                        f'.uid: {sub.actor.aid.uid}\n'
                         f'chan.raddr: {sub.chan.raddr}\n'
                     )
 
@@ -1150,7 +1150,7 @@ def test_peer_spawns_and_cancels_service_subactor(
 
                         assert isinstance(res, ContextCancelled)
                         assert client_ctx.cancel_acked
-                        assert res.canceller == root.uid
+                        assert res.canceller == root.aid.uid
                         assert not raise_sub_spawn_error_after
 
                         # cancelling the spawner sub should
@@ -1184,8 +1184,8 @@ def test_peer_spawns_and_cancels_service_subactor(
                         # little_bro: a `RuntimeError`.
                         #
                         check_inner_rte(rae)
-                        assert rae.relay_uid == client.chan.uid
-                        assert rae.src_uid == sub.chan.uid
+                        assert rae.relay_uid == client.chan.aid.uid
+                        assert rae.src_uid == sub.chan.aid.uid
 
                         assert not client_ctx.cancel_acked
                         assert (
@@ -1214,12 +1214,12 @@ def test_peer_spawns_and_cancels_service_subactor(
             except ContextCancelled as ctxc:
                 _ctxc = ctxc
                 print(
-                    f'{root.uid} caught ctxc from ctx with {client_ctx.chan.uid}\n'
+                    f'{root.aid.uid} caught ctxc from ctx with {client_ctx.chan.aid.uid}\n'
                     f'{repr(ctxc)}\n'
                 )
 
                 if not raise_sub_spawn_error_after:
-                    assert ctxc.canceller == root.uid
+                    assert ctxc.canceller == root.aid.uid
                 else:
                     assert ctxc.canceller == spawner_uid
 

tests/test_legacy_one_way_streaming.py

@@ -1,9 +1,11 @@
 """
-Streaming via async gen api
+Streaming via the, now legacy, "async-gen API".
+
 """
 import time
 from functools import partial
 import platform
+from typing import Callable
 
 import trio
 import tractor
@@ -19,7 +21,11 @@ def test_must_define_ctx():
         async def no_ctx():
             pass
 
-    assert "no_ctx must be `ctx: tractor.Context" in str(err.value)
+    assert (
+        "no_ctx must be `ctx: tractor.Context"
+        in
+        str(err.value)
+    )
 
     @tractor.stream
     async def has_ctx(ctx):
@@ -69,14 +75,14 @@ async def stream_from_single_subactor(
     async with tractor.open_nursery(
         registry_addrs=[reg_addr],
         start_method=start_method,
-    ) as nursery:
+    ) as an:
 
         async with tractor.find_actor('streamerd') as portals:
 
             if not portals:
 
                 # no brokerd actor found
-                portal = await nursery.start_actor(
+                portal = await an.start_actor(
                     'streamerd',
                     enable_modules=[__name__],
                 )
@@ -116,11 +122,22 @@ async def stream_from_single_subactor(
 
 
 @pytest.mark.parametrize(
-    'stream_func', [async_gen_stream, context_stream]
+    'stream_func',
+    [
+        async_gen_stream,
+        context_stream,
+    ],
+    ids='stream_func={}'.format
 )
-def test_stream_from_single_subactor(reg_addr, start_method, stream_func):
-    """Verify streaming from a spawned async generator.
-    """
+def test_stream_from_single_subactor(
+    reg_addr: tuple,
+    start_method: str,
+    stream_func: Callable,
+):
+    '''
+    Verify streaming from a spawned async generator.
+
+    '''
     trio.run(
         partial(
             stream_from_single_subactor,
@@ -132,10 +149,9 @@ def test_stream_from_single_subactor(reg_addr, start_method, stream_func):
 
 
 # this is the first 2 actors, streamer_1 and streamer_2
-async def stream_data(seed):
+async def stream_data(seed: int):
 
     for i in range(seed):
-
         yield i
 
         # trigger scheduler to simulate practical usage
@@ -143,15 +159,17 @@ async def stream_data(seed):
 
 
 # this is the third actor; the aggregator
-async def aggregate(seed):
-    """Ensure that the two streams we receive match but only stream
+async def aggregate(seed: int):
+    '''
+    Ensure that the two streams we receive match but only stream
     a single set of values to the parent.
-    """
-    async with tractor.open_nursery() as nursery:
+
+    '''
+    async with tractor.open_nursery() as an:
         portals = []
         for i in range(1, 3):
             # fork point
-            portal = await nursery.start_actor(
+            portal = await an.start_actor(
                 name=f'streamer_{i}',
                 enable_modules=[__name__],
             )
@@ -164,20 +182,28 @@ async def aggregate(seed):
             async with send_chan:
 
                 async with portal.open_stream_from(
-                    stream_data, seed=seed,
+                    stream_data,
+                    seed=seed,
                 ) as stream:
 
                     async for value in stream:
                         # leverage trio's built-in backpressure
                         await send_chan.send(value)
 
-            print(f"FINISHED ITERATING {portal.channel.uid}")
+            print(
+                f'FINISHED ITERATING!\n'
+                f'peer: {portal.channel.aid.uid}'
+            )
 
         # spawn 2 trio tasks to collect streams and push to a local queue
-        async with trio.open_nursery() as n:
+        async with trio.open_nursery() as tn:
 
             for portal in portals:
-                n.start_soon(push_to_chan, portal, send_chan.clone())
+                tn.start_soon(
+                    push_to_chan,
+                    portal,
+                    send_chan.clone(),
+                )
 
             # close this local task's reference to send side
             await send_chan.aclose()
@@ -194,20 +220,21 @@ async def aggregate(seed):
 
             print("FINISHED ITERATING in aggregator")
 
-        await nursery.cancel()
+        await an.cancel()
         print("WAITING on `ActorNursery` to finish")
     print("AGGREGATOR COMPLETE!")
 
 
-# this is the main actor and *arbiter*
-async def a_quadruple_example():
-    # a nursery which spawns "actors"
-    async with tractor.open_nursery() as nursery:
+async def a_quadruple_example() -> list[int]:
+    '''
+    Open the root-actor which is also a "registrar".
 
+    '''
+    async with tractor.open_nursery() as an:
         seed = int(1e3)
         pre_start = time.time()
 
-        portal = await nursery.start_actor(
+        portal = await an.start_actor(
             name='aggregator',
             enable_modules=[__name__],
         )
@@ -228,8 +255,14 @@ async def a_quadruple_example():
         return result_stream
 
 
-async def cancel_after(wait, reg_addr):
-    async with tractor.open_root_actor(registry_addrs=[reg_addr]):
+async def cancel_after(
+    wait: float,
+    reg_addr: tuple,
+) -> list[int]:
+
+    async with tractor.open_root_actor(
+        registry_addrs=[reg_addr],
+    ):
         with trio.move_on_after(wait):
             return await a_quadruple_example()
 
@@ -240,6 +273,10 @@ def time_quad_ex(
     ci_env: bool,
     spawn_backend: str,
 ):
+    non_linux: bool = (_sys := platform.system()) != 'Linux'
+    if ci_env and non_linux:
+        pytest.skip(f'Test is too flaky on {_sys!r} in CI')
+
     if spawn_backend == 'mp':
         '''
         no idea but the  mp *nix runs are flaking out here often...
@@ -247,16 +284,20 @@ def time_quad_ex(
         '''
         pytest.skip("Test is too flaky on mp in CI")
 
-    timeout = 7 if platform.system() in ('Windows', 'Darwin') else 4
+    timeout = 7 if non_linux else 4
     start = time.time()
-    results = trio.run(cancel_after, timeout, reg_addr)
-    diff = time.time() - start
+    results: list[int] = trio.run(
+        cancel_after,
+        timeout,
+        reg_addr,
+    )
+    diff: float = time.time() - start
     assert results
     return results, diff
 
 
 def test_a_quadruple_example(
-    time_quad_ex: tuple,
+    time_quad_ex: tuple[list[int], float],
     ci_env: bool,
     spawn_backend: str,
 ):
@@ -264,13 +305,12 @@ def test_a_quadruple_example(
     This also serves as a kind of "we'd like to be this fast test".
 
     '''
+    non_linux: bool = (_sys := platform.system()) != 'Linux'
+
     results, diff = time_quad_ex
     assert results
     this_fast = (
-        6 if platform.system() in (
-            'Windows',
-            'Darwin',
-        )
+        6 if non_linux
         else 3
     )
     assert diff < this_fast
@@ -281,19 +321,33 @@ def test_a_quadruple_example(
     list(map(lambda i: i/10, range(3, 9)))
 )
 def test_not_fast_enough_quad(
-    reg_addr, time_quad_ex, cancel_delay, ci_env, spawn_backend
+    reg_addr: tuple,
+    time_quad_ex: tuple[list[int], float],
+    cancel_delay: float,
+    ci_env: bool,
+    spawn_backend: str,
 ):
-    """Verify we can cancel midway through the quad example and all actors
-    cancel gracefully.
-    """
+    '''
+    Verify we can cancel midway through the quad example and all
+    actors cancel gracefully.
+
+    '''
     results, diff = time_quad_ex
     delay = max(diff - cancel_delay, 0)
-    results = trio.run(cancel_after, delay, reg_addr)
-    system = platform.system()
-    if system in ('Windows', 'Darwin') and results is not None:
+    results = trio.run(
+        cancel_after,
+        delay,
+        reg_addr,
+    )
+    system: str = platform.system()
+    if (
+        system in ('Windows', 'Darwin')
+        and
+        results is not None
+    ):
         # In CI envoirments it seems later runs are quicker then the first
         # so just ignore these
-        print(f"Woa there {system} caught your breath eh?")
+        print(f'Woa there {system} caught your breath eh?')
     else:
         # should be cancelled mid-streaming
         assert results is None
@@ -301,23 +355,24 @@ def test_not_fast_enough_quad(
 
 @tractor_test
 async def test_respawn_consumer_task(
-    reg_addr,
-    spawn_backend,
-    loglevel,
+    reg_addr: tuple,
+    spawn_backend: str,
+    loglevel: str,
 ):
-    """Verify that ``._portal.ReceiveStream.shield()``
+    '''
+    Verify that ``._portal.ReceiveStream.shield()``
     sucessfully protects the underlying IPC channel from being closed
     when cancelling and respawning a consumer task.
 
     This also serves to verify that all values from the stream can be
     received despite the respawns.
 
-    """
+    '''
     stream = None
 
-    async with tractor.open_nursery() as n:
+    async with tractor.open_nursery() as an:
 
-        portal = await n.start_actor(
+        portal = await an.start_actor(
             name='streamer',
             enable_modules=[__name__]
         )

tests/test_multi_program.py

@@ -35,6 +35,9 @@ if TYPE_CHECKING:
     )
 
 
+_non_linux: bool = platform.system() != 'Linux'
+
+
 def test_abort_on_sigint(
     daemon: subprocess.Popen,
 ):
@@ -137,6 +140,7 @@ def test_non_registrar_spawns_child(
     reg_addr: UnwrappedAddress,
     loglevel: str,
     debug_mode: bool,
+    ci_env: bool,
 ):
     '''
     Ensure a non-regristar (serving) root actor can spawn a sub and
@@ -148,6 +152,12 @@ def test_non_registrar_spawns_child(
 
     '''
     async def main():
+
+        # XXX, since apparently on macos in GH's CI it can be a race
+        # with the `daemon` registrar on grabbing the socket-addr..
+        if ci_env and _non_linux:
+            await trio.sleep(.5)
+
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
             loglevel=loglevel,

tests/test_root_infect_asyncio.py

@@ -49,7 +49,7 @@ def test_infected_root_actor(
                 ),
                 to_asyncio.open_channel_from(
                     aio_echo_server,
-                ) as (first, chan),
+                ) as (chan, first),
             ):
                 assert first == 'start'
 
@@ -91,13 +91,12 @@ def test_infected_root_actor(
 async def sync_and_err(
     # just signature placeholders for compat with
     # ``to_asyncio.open_channel_from()``
-    to_trio: trio.MemorySendChannel,
-    from_trio: asyncio.Queue,
+    chan: tractor.to_asyncio.LinkedTaskChannel,
     ev: asyncio.Event,
 
 ):
-    if to_trio:
-        to_trio.send_nowait('start')
+    if chan:
+        chan.started_nowait('start')
 
     await ev.wait()
     raise RuntimeError('asyncio-side')
@@ -174,7 +173,7 @@ def test_trio_prestarted_task_bubbles(
                             sync_and_err,
                             ev=aio_ev,
                         )
-                    ) as (first, chan),
+                    ) as (chan, first),
                 ):
 
                     for i in range(5):

tests/test_runtime.py

@@ -22,6 +22,10 @@ def unlink_file():
 async def crash_and_clean_tmpdir(
     tmp_file_path: str,
     error: bool = True,
+    rent_cancel: bool = True,
+
+    # XXX unused, but do we really need to test these cases?
+    self_cancel: bool = False,
 ):
     global _file_path
     _file_path = tmp_file_path
@@ -32,43 +36,75 @@ async def crash_and_clean_tmpdir(
     assert os.path.isfile(tmp_file_path)
     await trio.sleep(0.1)
     if error:
+        print('erroring in subactor!')
         assert 0
-    else:
+
+    elif self_cancel:
+        print('SELF-cancelling subactor!')
         actor.cancel_soon()
 
+    elif rent_cancel:
+        await trio.sleep_forever()
+
+    print('subactor exiting task!')
+
 
 @pytest.mark.parametrize(
     'error_in_child',
     [True, False],
+    ids='error_in_child={}'.format,
 )
 @tractor_test
 async def test_lifetime_stack_wipes_tmpfile(
     tmp_path,
     error_in_child: bool,
+    loglevel: str,
+    # log: tractor.log.StackLevelAdapter,
+    # ^TODO, once landed via macos support!
 ):
     child_tmp_file = tmp_path / "child.txt"
     child_tmp_file.touch()
     assert child_tmp_file.exists()
     path = str(child_tmp_file)
 
+    # NOTE, this is expected to cancel the sub
+    # in the `error_in_child=False` case!
+    timeout: float = (
+        1.6 if error_in_child
+        else 1
+    )
     try:
-        with trio.move_on_after(0.5):
-            async with tractor.open_nursery() as n:
-                    await (  # inlined portal
-                        await n.run_in_actor(
-                            crash_and_clean_tmpdir,
-                            tmp_file_path=path,
-                            error=error_in_child,
-                        )
-                    ).result()
-
+        with trio.move_on_after(timeout) as cs:
+            async with tractor.open_nursery(
+                loglevel=loglevel,
+            ) as an:
+                await (  # inlined `tractor.Portal`
+                    await an.run_in_actor(
+                        crash_and_clean_tmpdir,
+                        tmp_file_path=path,
+                        error=error_in_child,
+                    )
+                ).result()
     except (
         tractor.RemoteActorError,
-        # tractor.BaseExceptionGroup,
         BaseExceptionGroup,
-    ):
-        pass
+    ) as _exc:
+        exc = _exc
+        from tractor.log import get_console_log
+        log = get_console_log(
+            level=loglevel,
+            name=__name__,
+        )
+        log.exception(
+            f'Subactor failed as expected with {type(exc)!r}\n'
+        )
 
     # tmp file should have been wiped by
     # teardown stack.
     assert not child_tmp_file.exists()
+
+    if error_in_child:
+        assert not cs.cancel_called
+    else:
+        # expect timeout in some cases?
+        assert cs.cancel_called

tests/test_shm.py

@@ -2,6 +2,7 @@
 Shared mem primitives and APIs.
 
 """
+import platform
 import uuid
 
 # import numpy
@@ -53,7 +54,18 @@ def test_child_attaches_alot():
                     shm_key=shml.key,
                 ) as (ctx, start_val),
             ):
-                assert start_val == key
+                assert (_key := shml.key) == start_val
+
+                if platform.system() != 'Darwin':
+                    # XXX, macOS has a char limit..
+                    # see `ipc._shm._shorten_key_for_macos`
+                    assert (
+                        start_val
+                        ==
+                        key
+                        ==
+                        _key
+                    )
                 await ctx.result()
 
             await portal.cancel_actor()

tractor/_clustering.py

@@ -66,7 +66,7 @@ async def open_actor_cluster(
             trio.open_nursery() as tn,
             tractor.trionics.maybe_raise_from_masking_exc()
         ):
-            uid = tractor.current_actor().uid
+            uid = tractor.current_actor().aid.uid
 
             async def _start(name: str) -> None:
                 name = f'{uid[0]}.{name}'

tractor/_context.py

@@ -463,10 +463,11 @@ class Context:
 
     #     self._cancel_called = val
 
+    # TODO, use the `Actor.aid: Aid` instead!
     @property
     def canceller(self) -> tuple[str, str]|None:
         '''
-        `Actor.uid: tuple[str, str]` of the (remote)
+        `Actor.aid.uid: tuple[str, str]` of the (remote)
         actor-process who's task was cancelled thus causing this
         (side of the) context to also be cancelled.
 
@@ -499,12 +500,12 @@ class Context:
         if from_uid := re.src_uid:
             from_uid: tuple = tuple(from_uid)
 
-        our_uid: tuple = self._actor.uid
+        our_uid: tuple = self._actor.aid.uid
         our_canceller = self.canceller
 
         return bool(
             isinstance((ctxc := re), ContextCancelled)
-            and from_uid == self.chan.uid
+            and from_uid == self.chan.aid.uid
             and ctxc.canceller == our_uid
             and our_canceller == our_uid
         )
@@ -515,7 +516,7 @@ class Context:
         Records whether the task on the remote side of this IPC
         context acknowledged a cancel request via a relayed
         `ContextCancelled` with the `.canceller` attr set to the
-        `Actor.uid` of the local actor who's task entered
+        `Actor.aid.uid` of the local actor who's task entered
         `Portal.open_context()`.
 
         This will only be `True` when `.cancel()` is called and
@@ -789,8 +790,8 @@ class Context:
         # appropriately.
         log.runtime(
             'Setting remote error for ctx\n\n'
-            f'<= {self.peer_side!r}: {self.chan.uid}\n'
-            f'=> {self.side!r}: {self._actor.uid}\n\n'
+            f'<= {self.peer_side!r}: {self.chan.aid.reprol()}\n'
+            f'=> {self.side!r}: {self._actor.aid.reprol()}\n\n'
             f'{error!r}'
         )
         self._remote_error: BaseException = error
@@ -811,7 +812,7 @@ class Context:
             # cancelled.
             #
             # !TODO, switching to `Actor.aid` here!
-            if (canc := error.canceller) == self._actor.uid:
+            if (canc := error.canceller) == self._actor.aid.uid:
                 whom: str = 'us'
                 self._canceller = canc
             else:
@@ -1036,7 +1037,7 @@ class Context:
         ---------
         - after the far end cancels, the `.cancel()` calling side
           should receive a `ContextCancelled` with the
-          `.canceller: tuple` uid set to the current `Actor.uid`.
+          `.canceller: tuple` uid set to the current `Actor.aid.uid`.
 
         - timeout (quickly) on failure to rx this ACK error-msg in
           an attempt to sidestep 2-generals when the transport
@@ -1065,9 +1066,9 @@ class Context:
         )
         reminfo: str = (
             # ' =>\n'
-            # f'Context.cancel() => {self.chan.uid}\n'
+            # f'Context.cancel() => {self.chan.aid.uid}\n'
             f'\n'
-            f'c)=> {self.chan.uid}\n'
+            f'c)=> {self.chan.aid.reprol()}\n'
             f'   |_[{self.dst_maddr}\n'
             f'     >> {self.repr_rpc}\n'
             # f'    >> {self._nsf}() -> {codec}[dict]:\n\n'
@@ -1211,7 +1212,7 @@ class Context:
 
         '''
         __tracebackhide__: bool = hide_tb
-        peer_uid: tuple = self.chan.uid
+        peer_uid: tuple = self.chan.aid.uid
 
         # XXX NOTE XXX: `ContextCancelled`/`StreamOverrun` absorption
         # for "graceful cancellation" case(s):
@@ -1228,7 +1229,7 @@ class Context:
         # (`ContextCancelled`) as an expected
         # error-msg-is-cancellation-ack IFF said
         # `remote_error: ContextCancelled` has `.canceller`
-        # set to the `Actor.uid` of THIS task (i.e. the
+        # set to the `Actor.aid.uid` of THIS task (i.e. the
         # cancellation requesting task's actor is the actor
         # checking whether it should absorb the ctxc).
         self_ctxc: bool = self._is_self_cancelled(remote_error)
@@ -1679,7 +1680,7 @@ class Context:
 
         elif self._started_called:
             raise RuntimeError(
-                f'called `.started()` twice on context with {self.chan.uid}'
+                f'called `.started()` twice on context with {self.chan.aid.uid}'
             )
 
         started_msg = Started(
@@ -1812,7 +1813,7 @@ class Context:
         '''
         cid: str = self.cid
         chan: Channel = self.chan
-        from_uid: tuple[str, str]  = chan.uid
+        from_uid: tuple[str, str]  = chan.aid.uid
         send_chan: trio.MemorySendChannel = self._send_chan
         nsf: NamespacePath = self._nsf
 
@@ -1953,20 +1954,22 @@ class Context:
             # overrun state and that msg isn't stuck in an
             # overflow queue what happens?!?
 
-            local_uid = self._actor.uid
+            local_aid = self._actor.aid
             txt: str = (
                 'on IPC context:\n'
 
                 f'<= sender: {from_uid}\n'
                 f'  |_ {self._nsf}()\n\n'
 
-                f'=> overrun: {local_uid}\n'
+                f'=> overrun: {local_aid.reprol()!r}\n'
                 f'  |_cid: {cid}\n'
                 f'  |_task: {self._task}\n'
             )
             if not self._stream_opened:
                 txt += (
-                    f'\n*** No stream open on `{local_uid[0]}` side! ***\n\n'
+                    f'\n'
+                    f'*** No stream open on `{local_aid.name}` side! ***\n'
+                    f'\n'
                     f'{msg}\n'
                 )
 
@@ -2115,7 +2118,11 @@ async def open_context_from_portal(
     # XXX NOTE XXX: currenly we do NOT allow opening a contex
     # with "self" since the local feeder mem-chan processing
     # is not built for it.
-    if (uid := portal.channel.uid) == portal.actor.uid:
+    if (
+        (uid := portal.channel.aid.uid)
+        ==
+        portal.actor.aid.uid
+    ):
         raise RuntimeError(
             '** !! Invalid Operation !! **\n'
             'Can not open an IPC ctx with the local actor!\n'
@@ -2329,7 +2336,7 @@ async def open_context_from_portal(
             and
             ctxc is ctx._remote_error
             and
-            ctxc.canceller == portal.actor.uid
+            ctxc.canceller == portal.actor.aid.uid
         ):
             log.cancel(
                 f'Context (cid=[{ctx.cid[-6:]}..] cancelled gracefully with:\n'
@@ -2406,7 +2413,7 @@ async def open_context_from_portal(
         logmeth(msg)
 
         if debug_mode():
-            # async with debug.acquire_debug_lock(portal.actor.uid):
+            # async with debug.acquire_debug_lock(portal.actor.aid.uid):
             #     pass
             # TODO: factor ^ into below for non-root cases?
             #

tractor/_exceptions.py

@@ -982,6 +982,7 @@ class TransportClosed(Exception):
         '''
         __tracebackhide__: bool = hide_tb
         message: str = message or self.message
+
         # when a cause is set, slap it onto the log emission.
         if cause := self.src_exc:
             cause_tb_str: str = ''.join(
@@ -989,7 +990,7 @@ class TransportClosed(Exception):
             )
             message += (
                 f'{cause_tb_str}\n'  # tb
-                f'    {cause}\n'  # exc repr
+                f'{cause!r}\n'  # exc repr
             )
 
         getattr(

tractor/_rpc.py

@@ -252,8 +252,8 @@ async def _invoke_non_context(
                 ):
                     log.warning(
                         'Failed to send RPC result?\n'
-                        f'|_{func}@{actor.uid}() -> {ret_msg}\n\n'
-                        f'x=> peer: {chan.uid}\n'
+                        f'|_{func}@{actor.aid.reprol()}() -> {ret_msg}\n\n'
+                        f'x=> peer: {chan.aid.reprol()}\n'
                     )
 
 @acm
@@ -698,7 +698,7 @@ async def _invoke(
             #   which cancels the scope presuming the input error
             #   is not a `.cancel_acked` pleaser.
             if rpc_ctx_cs.cancelled_caught:
-                our_uid: tuple = actor.uid
+                our_uid: tuple = actor.aid.uid
 
                 # first check for and raise any remote error
                 # before raising any context cancelled case
@@ -730,7 +730,7 @@ async def _invoke(
                     # TODO: determine if the ctx peer task was the
                     # exact task which cancelled, vs. some other
                     # task in the same actor.
-                    elif canceller == ctx.chan.uid:
+                    elif canceller == ctx.chan.aid.uid:
                         explain += f'its {ctx.peer_side!r}-side peer'
 
                     elif canceller == our_uid:
@@ -825,7 +825,7 @@ async def _invoke(
             # associated child isn't in debug any more
             await debug.maybe_wait_for_debugger()
             ctx: Context = actor._contexts.pop((
-                chan.uid,
+                chan.aid.uid,
                 cid,
             ))
 
@@ -927,7 +927,7 @@ async def try_ship_error_to_remote(
             log.critical(
                 'IPC transport failure -> '
                 f'failed to ship error to {remote_descr}!\n\n'
-                f'{type(msg)!r}[{msg.boxed_type_str}] X=> {channel.uid}\n'
+                f'{type(msg)!r}[{msg.boxed_type_str}] X=> {channel.aid.uid}\n'
                 f'\n'
                 # TODO: use `.msg.preetty_struct` for this!
                 f'{msg}\n'
@@ -1005,7 +1005,7 @@ async def process_messages(
             async for msg in chan:
                 log.transport(   # type: ignore
                     f'IPC msg from peer\n'
-                    f'<= {chan.uid}\n\n'
+                    f'<= {chan.aid.reprol()}\n\n'
 
                     # TODO: use of the pprinting of structs is
                     # FRAGILE and should prolly not be
@@ -1109,7 +1109,7 @@ async def process_messages(
                         except BaseException:
                             log.exception(
                                 'Failed to cancel task?\n'
-                                f'<= canceller: {chan.uid}\n'
+                                f'<= canceller: {chan.aid.reprol()}\n'
                                 f'  |_{chan}\n\n'
                                 f'=> {actor}\n'
                                 f'  |_cid: {target_cid}\n'
@@ -1264,7 +1264,7 @@ async def process_messages(
 
                 log.transport(
                     'Waiting on next IPC msg from\n'
-                    f'peer: {chan.uid}\n'
+                    f'peer: {chan.aid.reprol()}\n'
                     f'|_{chan}\n'
                 )
 
@@ -1313,12 +1313,10 @@ async def process_messages(
                 f'peer IPC channel closed abruptly?\n'
                 f'\n'
                 f'<=x[\n'
-                f'  {chan}\n'
-                f'  |_{chan.raddr}\n\n'
+                f'{chan}\n'
             )
             +
             tc.message
-
         )
 
         # transport **WAS** disconnected
@@ -1341,8 +1339,8 @@ async def process_messages(
             match err:
                 case ContextCancelled():
                     log.cancel(
-                        f'Actor: {actor.uid} was context-cancelled with,\n'
-                        f'str(err)'
+                        f'Actor: {actor.aid.reprol()!r} is ctxc with,\n'
+                        f'{str(err)}'
                     )
                 case _:
                     log.exception("Actor errored:")

tractor/_runtime.py

@@ -691,7 +691,7 @@ class Actor:
 
         '''
         # ?TODO, use Aid here as well?
-        actor_uid = chan.uid
+        actor_uid = chan.aid.uid
         assert actor_uid
         try:
             ctx = self._contexts[(
@@ -701,7 +701,7 @@ class Actor:
             )]
             log.debug(
                 f'Retreived cached IPC ctx for\n'
-                f'peer: {chan.uid}\n'
+                f'peer: {chan.aid.uid}\n'
                 f'cid:{cid}\n'
             )
             ctx._allow_overruns: bool = allow_overruns
@@ -718,7 +718,7 @@ class Actor:
         except KeyError:
             log.debug(
                 f'Allocate new IPC ctx for\n'
-                f'peer: {chan.uid}\n'
+                f'peer: {chan.aid.uid}\n'
                 f'cid: {cid}\n'
             )
             ctx = mk_context(
@@ -764,7 +764,7 @@ class Actor:
 
         '''
         cid: str = str(uuid.uuid4())
-        assert chan.uid
+        assert chan.aid.uid
         ctx = self.get_context(
             chan=chan,
             cid=cid,
@@ -791,12 +791,12 @@ class Actor:
             ns=ns,
             func=func,
             kwargs=kwargs,
-            uid=self.uid,
+            uid=self.aid.uid,  # <- !TODO use .aid!
             cid=cid,
         )
         log.runtime(
             'Sending RPC `Start`\n\n'
-            f'=> peer: {chan.uid}\n'
+            f'=> peer: {chan.aid.uid}\n'
             f'  |_ {ns}.{func}({kwargs})\n\n'
 
             f'{pretty_struct.pformat(msg)}'
@@ -1244,7 +1244,7 @@ class Actor:
                 'Cancel request for invalid RPC task.\n'
                 'The task likely already completed or was never started!\n\n'
                 f'<= canceller: {requesting_aid}\n'
-                f'=> {cid}@{parent_chan.uid}\n'
+                f'=> {cid}@{parent_chan.aid.uid}\n'
                 f'  |_{parent_chan}\n'
             )
             return True
@@ -1381,7 +1381,7 @@ class Actor:
             f'Cancelling {descr} RPC tasks\n\n'
             f'<=c) {req_aid} [canceller]\n'
             f'{rent_chan_repr}'
-            f'c)=> {self.uid} [cancellee]\n'
+            f'c)=> {self.aid.uid} [cancellee]\n'
             f'  |_{self} [with {len(tasks)} tasks]\n'
             # f'  |_tasks: {len(tasks)}\n'
             # f'{tasks_str}'
@@ -1687,7 +1687,7 @@ async def async_main(
                             await reg_portal.run_from_ns(
                                 'self',
                                 'register_actor',
-                                uid=actor.uid,
+                                uid=actor.aid.uid,
                                 addr=accept_addr.unwrap(),
                             )
 
@@ -1758,9 +1758,11 @@ async def async_main(
         # always!
         match internal_err:
             case ContextCancelled():
+                reprol: str = actor.aid.reprol()
                 log.cancel(
-                    f'Actor: {actor.uid} was task-context-cancelled with,\n'
-                    f'str(internal_err)'
+                    f'Actor {reprol!r} was task-ctx-cancelled with,\n'
+                    f'\n'
+                    f'{internal_err!r}'
                 )
             case _:
                 log.exception(
@@ -1832,7 +1834,7 @@ async def async_main(
                             await reg_portal.run_from_ns(
                                 'self',
                                 'unregister_actor',
-                                uid=actor.uid
+                                uid=actor.aid.uid,
                             )
                     except OSError:
                         failed = True

tractor/_spawn.py

@@ -151,7 +151,7 @@ async def exhaust_portal(
     __tracebackhide__ = True
     try:
         log.debug(
-            f'Waiting on final result from {actor.uid}'
+            f'Waiting on final result from {actor.aid.uid}'
         )
 
         # XXX: streams should never be reaped here since they should
@@ -210,17 +210,17 @@ async def cancel_on_completion(
         actor,
     )
     if isinstance(result, Exception):
-        errors[actor.uid]: Exception = result
+        errors[actor.aid.uid]: Exception = result
         log.cancel(
             'Cancelling subactor runtime due to error:\n\n'
-            f'Portal.cancel_actor() => {portal.channel.uid}\n\n'
+            f'Portal.cancel_actor() => {portal.channel.aid}\n\n'
             f'error: {result}\n'
         )
 
     else:
         log.runtime(
             'Cancelling subactor gracefully:\n\n'
-            f'Portal.cancel_actor() => {portal.channel.uid}\n\n'
+            f'Portal.cancel_actor() => {portal.channel.aid}\n\n'
             f'result: {result}\n'
         )
 
@@ -308,7 +308,7 @@ async def hard_kill(
         #     )
         #     with trio.CancelScope(shield=True):
         #         async with debug.acquire_debug_lock(
-        #             subactor_uid=current_actor().uid,
+        #             subactor_uid=current_actor().aid.uid,
         #         ) as _ctx:
         #             log.warning(
         #                 'Acquired debug lock, child ready to be killed ??\n'
@@ -483,7 +483,7 @@ async def trio_proc(
         # TODO, how to pass this over "wire" encodings like
         # cmdline args?
         # -[ ] maybe we can add an `msgtypes.Aid.min_tuple()` ?
-        str(subactor.uid),
+        str(subactor.aid.uid),
         # Address the child must connect to on startup
         "--parent_addr",
         str(parent_addr)
@@ -514,7 +514,7 @@ async def trio_proc(
             # channel should have handshake completed by the
             # local actor by the time we get a ref to it
             event, chan = await ipc_server.wait_for_peer(
-                subactor.uid
+                subactor.aid.uid
             )
 
         except trio.Cancelled:
@@ -528,7 +528,9 @@ async def trio_proc(
                         await debug.maybe_wait_for_debugger()
 
                     elif proc is not None:
-                        async with debug.acquire_debug_lock(subactor.uid):
+                        async with debug.acquire_debug_lock(
+                            subactor_uid=subactor.aid.uid
+                        ):
                             # soft wait on the proc to terminate
                             with trio.move_on_after(0.5):
                                 await proc.wait()
@@ -538,7 +540,7 @@ async def trio_proc(
         assert proc
 
         portal = Portal(chan)
-        actor_nursery._children[subactor.uid] = (
+        actor_nursery._children[subactor.aid.uid] = (
             subactor,
             proc,
             portal,
@@ -563,7 +565,7 @@ async def trio_proc(
 
         # track subactor in current nursery
         curr_actor: Actor = current_actor()
-        curr_actor._actoruid2nursery[subactor.uid] = actor_nursery
+        curr_actor._actoruid2nursery[subactor.aid.uid] = actor_nursery
 
         # resume caller at next checkpoint now that child is up
         task_status.started(portal)
@@ -616,7 +618,9 @@ async def trio_proc(
                 # 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 debug.acquire_debug_lock(
+                        subactor_uid=subactor.aid.uid
+                    ):
                         with trio.move_on_after(0.5):
                             await proc.wait()
 
@@ -662,7 +666,7 @@ async def trio_proc(
         if not cancelled_during_spawn:
             # pop child entry to indicate we no longer managing this
             # subactor
-            actor_nursery._children.pop(subactor.uid)
+            actor_nursery._children.pop(subactor.aid.uid)
 
 
 async def mp_proc(
@@ -744,7 +748,7 @@ async def mp_proc(
     # register the process before start in case we get a cancel
     # request before the actor has fully spawned - then we can wait
     # for it to fully come up before sending a cancel request
-    actor_nursery._children[subactor.uid] = (subactor, proc, None)
+    actor_nursery._children[subactor.aid.uid] = (subactor, proc, None)
 
     proc.start()
     if not proc.is_alive():
@@ -758,7 +762,7 @@ async def mp_proc(
         # channel should have handshake completed by the
         # local actor by the time we get a ref to it
         event, chan = await ipc_server.wait_for_peer(
-            subactor.uid,
+            subactor.aid.uid,
         )
 
         # XXX: monkey patch poll API to match the ``subprocess`` API..
@@ -771,7 +775,7 @@ async def mp_proc(
         # any process we may have started.
 
         portal = Portal(chan)
-        actor_nursery._children[subactor.uid] = (subactor, proc, portal)
+        actor_nursery._children[subactor.aid.uid] = (subactor, proc, portal)
 
         # unblock parent task
         task_status.started(portal)
@@ -810,7 +814,7 @@ async def mp_proc(
             # tandem if not done already
             log.warning(
                 "Cancelling existing result waiter task for "
-                f"{subactor.uid}")
+                f"{subactor.aid.uid}")
             nursery.cancel_scope.cancel()
 
     finally:
@@ -828,7 +832,7 @@ async def mp_proc(
         log.debug(f"Joined {proc}")
 
         # pop child entry to indicate we are no longer managing subactor
-        actor_nursery._children.pop(subactor.uid)
+        actor_nursery._children.pop(subactor.aid.uid)
 
         # TODO: prolly report to ``mypy`` how this causes all sorts of
         # false errors..

tractor/_state.py

@@ -104,7 +104,7 @@ def current_actor(
             msg += (
                 f'Apparently the lact active actor was\n'
                 f'|_{last}\n'
-                f'|_{last.uid}\n'
+                f'|_{last.aid.uid}\n'
             )
         # no actor runtime has (as of yet) ever been started for
         # this process.
@@ -175,27 +175,51 @@ def current_ipc_ctx(
 
 def get_rt_dir(
     subdir: str|Path|None = None,
+    appname: str = 'tractor',
 ) -> Path:
     '''
     Return the user "runtime dir", the file-sys location where most
     userspace apps stick their IPC and cache related system
     util-files.
 
-    On linux we take use a `'${XDG_RUNTIME_DIR}/tractor/` subdir by
-    default but equivalents are mapped for each platform using
-    the lovely `platformdirs`.
+    On linux we use a `${XDG_RUNTIME_DIR}/tractor/` subdir by
+    default, but equivalents are mapped for each platform using
+    the lovely `platformdirs` lib.
 
     '''
     rt_dir: Path = Path(
         platformdirs.user_runtime_dir(
-            appname='tractor',
+            appname=appname,
         ),
     )
+
+    # Normalize and validate that `subdir` is a relative path
+    # without any parent-directory ("..") components, to prevent
+    # escaping the runtime directory.
     if subdir:
-        rt_dir: Path = rt_dir / subdir
+        subdir_path = (
+            subdir
+            if isinstance(subdir, Path)
+            else Path(subdir)
+        )
+        if subdir_path.is_absolute():
+            raise ValueError(
+                f'`subdir` must be a relative path!\n'
+                f'{subdir!r}\n'
+            )
+        if any(part == '..' for part in subdir_path.parts):
+            raise ValueError(
+                "`subdir` must not contain '..' components!\n"
+                f'{subdir!r}\n'
+            )
+
+        rt_dir: Path = rt_dir / subdir_path
 
     if not rt_dir.is_dir():
-        rt_dir.mkdir()
+        rt_dir.mkdir(
+            parents=True,
+            exist_ok=True,  # avoid `FileExistsError` from conc calls
+        )
 
     return rt_dir
 

tractor/_supervise.py

@@ -391,15 +391,17 @@ class ActorNursery:
 
                     else:
                         if portal is None:  # actor hasn't fully spawned yet
-                            event: trio.Event = server._peer_connected[subactor.uid]
+                            event: trio.Event = server._peer_connected[
+                                subactor.aid.uid
+                            ]
                             log.warning(
-                                f"{subactor.uid} never 't finished spawning?"
+                                f"{subactor.aid.uid} never 't finished spawning?"
                             )
 
                             await event.wait()
 
                             # channel/portal should now be up
-                            _, _, portal = children[subactor.uid]
+                            _, _, portal = children[subactor.aid.uid]
 
                             # XXX should be impossible to get here
                             # unless method was called from within
@@ -407,7 +409,7 @@ class ActorNursery:
                             if portal is None:
                                 # cancelled while waiting on the event
                                 # to arrive
-                                chan = server._peers[subactor.uid][-1]
+                                chan = server._peers[subactor.aid.uid][-1]
                                 if chan:
                                     portal = Portal(chan)
                                 else:  # there's no other choice left
@@ -506,7 +508,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
 
                 except BaseException as _inner_err:
                     inner_err = _inner_err
-                    errors[actor.uid] = inner_err
+                    errors[actor.aid.uid] = inner_err
 
                     # If we error in the root but the debugger is
                     # engaged we don't want to prematurely kill (and
@@ -539,7 +541,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
                             log.cancel(
                                 f'Actor-nursery cancelled by {etype}\n\n'
 
-                                f'{current_actor().uid}\n'
+                                f'{current_actor().aid.uid}\n'
                                 f' |_{an}\n\n'
 
                                 # TODO: show tb str?
@@ -630,7 +632,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
 
             # show frame on any (likely) internal error
             if (
-                not an.cancelled
+                not an.cancel_called
                 and an._scope_error
             ):
                 __tracebackhide__: bool = False
@@ -726,7 +728,7 @@ async def open_nursery(
         if (
             an
             and
-            not an.cancelled
+            not an.cancel_called
             and
             an._scope_error
         ):

tractor/_testing/addr.py

@@ -61,7 +61,11 @@ def get_rando_addr(
         # NOTE, file-name uniqueness (no-collisions) will be based on
         # the runtime-directory and root (pytest-proc's) pid.
         case 'uds':
-            testrun_reg_addr = addr_type.get_random().unwrap()
+            from tractor.ipc._uds import UDSAddress
+            addr: UDSAddress = addr_type.get_random()
+            assert addr.is_valid
+            assert addr.sockpath.resolve()
+            testrun_reg_addr = addr.unwrap()
 
     # XXX, as sanity it should never the same as the default for the
     # host-singleton registry actor.

tractor/_testing/pytest.py

@@ -74,6 +74,7 @@ def tractor_test(fn):
         reg_addr=None,
         start_method: str|None = None,
         debug_mode: bool = False,
+        tpt_proto: str|None=None,
         **kwargs
     ):
         # __tracebackhide__ = True
@@ -102,6 +103,9 @@ def tractor_test(fn):
             # set of subprocess spawning backends
             kwargs['debug_mode'] = debug_mode
 
+        if 'tpt_proto' in inspect.signature(fn).parameters:
+            # set of subprocess spawning backends
+            kwargs['tpt_proto'] = tpt_proto
 
         if kwargs:
 
@@ -177,6 +181,13 @@ def pytest_configure(config):
     backend = config.option.spawn_backend
     tractor._spawn.try_set_start_method(backend)
 
+    # register custom marks to avoid warnings see,
+    # https://docs.pytest.org/en/stable/how-to/writing_plugins.html#registering-custom-markers
+    config.addinivalue_line(
+        'markers',
+        'no_tpt(proto_key): test will (likely) not behave with tpt backend'
+    )
+
 
 @pytest.fixture(scope='session')
 def debug_mode(request) -> bool:
@@ -225,13 +236,32 @@ def tpt_protos(request) -> list[str]:
     autouse=True,
 )
 def tpt_proto(
+    request,
     tpt_protos: list[str],
 ) -> str:
     proto_key: str = tpt_protos[0]
 
+    # ?TODO, but needs a fn-scoped tpt_proto fixture..
+    # @pytest.mark.no_tpt('uds')
+    # node = request.node
+    # markers = node.own_markers
+    # for mark in markers:
+    #     if (
+    #         mark.name == 'no_tpt'
+    #         and
+    #         proto_key in mark.args
+    #     ):
+    #         pytest.skip(
+    #             f'Test {node} normally fails with '
+    #             f'tpt-proto={proto_key!r}\n'
+    #         )
+
     from tractor import _state
     if _state._def_tpt_proto != proto_key:
         _state._def_tpt_proto = proto_key
+        _state._runtime_vars['_enable_tpts'] = [
+            proto_key,
+        ]
 
     yield proto_key
 

tractor/devx/debug/_sigint.py

@@ -21,6 +21,7 @@ cancellation during REPL interaction.
 
 '''
 from __future__ import annotations
+import platform
 from typing import (
     TYPE_CHECKING,
 )
@@ -49,6 +50,7 @@ if TYPE_CHECKING:
 
 log = get_logger()
 
+_is_macos: bool = platform.system() == 'Darwin'
 _ctlc_ignore_header: str = (
     'Ignoring SIGINT while debug REPL in use'
 )
@@ -300,6 +302,11 @@ def sigint_shield(
         # XXX: yah, mega hack, but how else do we catch this madness XD
         if (
             repl.shname == 'xonsh'
+            or (
+                repl.shname == 'bash'
+                and
+                _is_macos
+            )
         ):
             flush_status += (
                 '-> ALSO re-flushing due to `xonsh`..\n'

tractor/ipc/_chan.py

@@ -94,7 +94,7 @@ class Channel:
         self._transport: MsgTransport|None = transport
 
         # set after handshake - always info from peer end
-        self.aid: Aid|None = None
+        self._aid: Aid|None = None
 
         self._aiter_msgs = self._iter_msgs()
         self._exc: Exception|None = None
@@ -122,6 +122,14 @@ class Channel:
         '''
         return self._cancel_called
 
+    @property
+    def aid(self) -> Aid:
+        '''
+        Peer actor's ID.
+
+        '''
+        return self._aid
+
     @property
     def uid(self) -> tuple[str, str]:
         '''
@@ -505,7 +513,7 @@ class Channel:
             f'<= {peer_aid.reprol(sin_uuid=False)}\n'
         )
         # NOTE, we always are referencing the remote peer!
-        self.aid = peer_aid
+        self._aid = peer_aid
         return peer_aid
 
 

tractor/ipc/_shm.py

@@ -23,6 +23,7 @@ considered optional within the context of this runtime-library.
 
 """
 from __future__ import annotations
+import hashlib
 from multiprocessing import shared_memory as shm
 from multiprocessing.shared_memory import (
     # SharedMemory,
@@ -106,11 +107,12 @@ class NDToken(Struct, frozen=True):
     This type is msg safe.
 
     '''
-    shm_name: str  # this servers as a "key" value
+    shm_name: str  # actual OS-level name (may be shortened on macOS)
     shm_first_index_name: str
     shm_last_index_name: str
     dtype_descr: tuple
     size: int  # in struct-array index / row terms
+    key: str|None = None  # original descriptive key (for lookup)
 
     # TODO: use nptyping here on dtypes
     @property
@@ -124,6 +126,41 @@ class NDToken(Struct, frozen=True):
     def as_msg(self):
         return to_builtins(self)
 
+    def __eq__(self, other) -> bool:
+        '''
+        Compare tokens based on shm names and dtype,
+        ignoring the `key` field.
+
+        The `key` field is only used for lookups,
+        not for token identity.
+
+        '''
+        if not isinstance(other, NDToken):
+            return False
+        return (
+            self.shm_name == other.shm_name
+            and self.shm_first_index_name
+                == other.shm_first_index_name
+            and self.shm_last_index_name
+                == other.shm_last_index_name
+            and self.dtype_descr == other.dtype_descr
+            and self.size == other.size
+        )
+
+    def __hash__(self) -> int:
+        '''
+        Hash based on the same fields used
+        in `.__eq__()`.
+
+        '''
+        return hash((
+            self.shm_name,
+            self.shm_first_index_name,
+            self.shm_last_index_name,
+            self.dtype_descr,
+            self.size,
+        ))
+
     @classmethod
     def from_msg(cls, msg: dict) -> NDToken:
         if isinstance(msg, NDToken):
@@ -160,6 +197,50 @@ def get_shm_token(key: str) -> NDToken | None:
     return _known_tokens.get(key)
 
 
+def _shorten_key_for_macos(
+    key: str,
+    prefix: str = '',
+    suffix: str = '',
+) -> str:
+    '''
+    MacOS has a (hillarious) 31 character limit for POSIX shared
+    memory names. Hash long keys to fit within this limit while
+    maintaining uniqueness.
+
+    '''
+    # macOS shm_open() has a 31 char limit (PSHMNAMLEN)
+    # format: /t_<hash16> = 19 chars, well under limit
+    max_len: int = 31
+    if len(key) <= max_len:
+        return key
+
+    _hash: str = hashlib.sha256(
+        key.encode()
+    ).hexdigest()
+
+    hash_len: int = (
+        (max_len - 1)
+        - len(prefix)
+        - len(suffix)
+    )
+    key_hash: str = _hash[:hash_len]
+    short_key = (
+        prefix
+        +
+        f'{key_hash}'
+        +
+        suffix
+    )
+
+    log.debug(
+        f'Shortened shm key for macOS:\n'
+        f'  original: {key!r} ({len(key)!r} chars)\n'
+        f'  shortened: {short_key!r}'
+        f' ({len(short_key)!r} chars)'
+    )
+    return short_key
+
+
 def _make_token(
     key: str,
     size: int,
@@ -171,12 +252,32 @@ def _make_token(
     to access a shared array.
 
     '''
+    # On macOS, shorten keys that exceed the
+    # 31 character limit
+    if platform.system() == 'Darwin':
+        shm_name = _shorten_key_for_macos(
+            key=key,
+        )
+        shm_first = _shorten_key_for_macos(
+            key=key,
+            suffix='_first',
+        )
+        shm_last = _shorten_key_for_macos(
+            key=key,
+            suffix='_last',
+        )
+    else:
+        shm_name = key
+        shm_first = key + '_first'
+        shm_last = key + '_last'
+
     return NDToken(
-        shm_name=key,
-        shm_first_index_name=key + "_first",
-        shm_last_index_name=key + "_last",
+        shm_name=shm_name,
+        shm_first_index_name=shm_first,
+        shm_last_index_name=shm_last,
         dtype_descr=tuple(np.dtype(dtype).descr),
         size=size,
+        key=key,  # store original key for lookup
     )
 
 
@@ -431,9 +532,17 @@ class ShmArray:
 
     def destroy(self) -> None:
         if _USE_POSIX:
-            # We manually unlink to bypass all the "resource tracker"
-            # nonsense meant for non-SC systems.
-            shm_unlink(self._shm.name)
+            # We manually unlink to bypass all the
+            # "resource tracker" nonsense meant for
+            # non-SC systems.
+            name = self._shm.name
+            try:
+                shm_unlink(name)
+            except FileNotFoundError:
+                # might be a teardown race here?
+                log.warning(
+                    f'Shm for {name} already unlinked?'
+                )
 
         self._first.destroy()
         self._last.destroy()
@@ -463,8 +572,16 @@ def open_shm_ndarray(
     a = np.zeros(size, dtype=dtype)
     a['index'] = np.arange(len(a))
 
+    # Create token first to get the (possibly
+    # shortened) shm name
+    token = _make_token(
+        key=key,
+        size=size,
+        dtype=dtype,
+    )
+
     shm = SharedMemory(
-        name=key,
+        name=token.shm_name,
         create=True,
         size=a.nbytes
     )
@@ -476,12 +593,6 @@ def open_shm_ndarray(
     array[:] = a[:]
     array.setflags(write=int(not readonly))
 
-    token = _make_token(
-        key=key,
-        size=size,
-        dtype=dtype,
-    )
-
     # create single entry arrays for storing an first and last indices
     first = SharedInt(
         shm=SharedMemory(
@@ -554,13 +665,23 @@ def attach_shm_ndarray(
 
     '''
     token = NDToken.from_msg(token)
-    key = token.shm_name
+    # Use original key for _known_tokens lookup,
+    # shm_name for OS calls
+    lookup_key = (
+        token.key if token.key
+        else token.shm_name
+    )
 
-    if key in _known_tokens:
-        assert NDToken.from_msg(_known_tokens[key]) == token, "WTF"
+    if lookup_key in _known_tokens:
+        assert (
+            NDToken.from_msg(
+                _known_tokens[lookup_key]
+            ) == token
+        ), 'WTF'
 
-    # XXX: ugh, looks like due to the ``shm_open()`` C api we can't
-    # actually place files in a subdir, see discussion here:
+    # XXX: ugh, looks like due to the ``shm_open()``
+    # C api we can't actually place files in a subdir,
+    # see discussion here:
     # https://stackoverflow.com/a/11103289
 
     # attach to array buffer and view as per dtype
@@ -568,7 +689,7 @@ def attach_shm_ndarray(
     for _ in range(3):
         try:
             shm = SharedMemory(
-                name=key,
+                name=token.shm_name,
                 create=False,
             )
             break
@@ -614,10 +735,10 @@ def attach_shm_ndarray(
     sha.array
 
     # Stash key -> token knowledge for future queries
-    # via `maybe_opepn_shm_array()` but only after we know
-    # we can attach.
-    if key not in _known_tokens:
-        _known_tokens[key] = token
+    # via `maybe_open_shm_ndarray()` but only after
+    # we know we can attach.
+    if lookup_key not in _known_tokens:
+        _known_tokens[lookup_key] = token
 
     # "close" attached shm on actor teardown
     tractor.current_actor().lifetime_stack.callback(sha.close)
@@ -661,7 +782,10 @@ def maybe_open_shm_ndarray(
             False,  # not newly opened
         )
     except KeyError:
-        log.warning(f"Could not find {key} in shms cache")
+        log.warning(
+            f'Could not find key in shms cache,\n'
+            f'key: {key!r}\n'
+        )
         if dtype:
             token = _make_token(
                 key,
@@ -771,6 +895,7 @@ def open_shm_list(
     size: int = int(2 ** 10),
     dtype: float | int | bool | str | bytes | None = float,
     readonly: bool = True,
+    prefix: str = 'shml_',
 
 ) -> ShmList:
 
@@ -784,6 +909,12 @@ def open_shm_list(
         }[dtype]
         sequence = [default] * size
 
+    if platform.system() == 'Darwin':
+        key: str = _shorten_key_for_macos(
+            key=key,
+            prefix=prefix,
+        )
+
     shml = ShmList(
         sequence=sequence,
         name=key,

tractor/ipc/_uds.py

@@ -23,12 +23,12 @@ from contextlib import (
 )
 from pathlib import Path
 import os
+import sys
 from socket import (
     AF_UNIX,
     SOCK_STREAM,
-    SO_PASSCRED,
-    SO_PEERCRED,
     SOL_SOCKET,
+    error as socket_error,
 )
 import struct
 from typing import (
@@ -53,7 +53,7 @@ from tractor.log import get_logger
 from tractor.ipc._transport import (
     MsgpackTransport,
 )
-from .._state import (
+from tractor._state import (
     get_rt_dir,
     current_actor,
     is_root_process,
@@ -63,6 +63,28 @@ if TYPE_CHECKING:
     from ._runtime import Actor
 
 
+# Platform-specific credential passing constants
+# See: https://stackoverflow.com/a/7982749
+if sys.platform == 'linux':
+    from socket import (
+        SO_PASSCRED,
+        SO_PEERCRED,
+    )
+
+else:
+    # Other (Unix) platforms - though further testing is required and
+    # others may need additional special handling?
+    SO_PASSCRED = None
+    SO_PEERCRED = None
+
+    # NOTE, macOS uses `LOCAL_PEERCRED` instead of `SO_PEERCRED` and
+    # doesn't need `SO_PASSCRED` (credential passing is always enabled).
+    # See code in <sys/un.h>: `#define LOCAL_PEERCRED 0x001`
+    #
+    # XXX INSTEAD we use the (hopefully) more generic
+    # `get_peer_pid()` below for other OSes.
+
+
 log = get_logger()
 
 
@@ -165,15 +187,21 @@ class UDSAddress(
             err_on_no_runtime=False,
         )
         if actor:
-            sockname: str = '::'.join(actor.uid) + f'@{pid}'
+            sockname: str = f'{actor.aid.name}@{pid}'
+            # XXX, orig version which broke both macOS (file-name
+            # length) and `multiaddrs` ('::' invalid separator).
+            # sockname: str = '::'.join(actor.uid) + f'@{pid}'
+            #
             # ?^TODO, for `multiaddr`'s parser we can't use the `::`
             # above^, SO maybe a `.` or something else here?
             # sockname: str = '.'.join(actor.uid) + f'@{pid}'
             # -[ ] CURRENTLY using `.` BREAKS TEST SUITE tho..
         else:
-            prefix: str = '<unknown-actor>'
             if is_root_process():
-                prefix: str = 'root'
+                prefix: str = 'no_runtime_root'
+            else:
+                prefix: str = 'no_runtime_actor'
+
             sockname: str = f'{prefix}@{pid}'
 
         sockpath: Path = Path(f'{sockname}.sock')
@@ -292,7 +320,12 @@ def close_listener(
 
 
 async def open_unix_socket_w_passcred(
-    filename: str|bytes|os.PathLike[str]|os.PathLike[bytes],
+    filename: (
+        str
+        |bytes
+        |os.PathLike[str]
+        |os.PathLike[bytes]
+    ),
 ) -> trio.SocketStream:
     '''
     Literally the exact same as `trio.open_unix_socket()` except we set the additiona
@@ -310,21 +343,66 @@ async def open_unix_socket_w_passcred(
     # much more simplified logic vs tcp sockets - one socket type and only one
     # possible location to connect to
     sock = trio.socket.socket(AF_UNIX, SOCK_STREAM)
-    sock.setsockopt(SOL_SOCKET, SO_PASSCRED, 1)
+
+    # Only set SO_PASSCRED on Linux (not needed/available on macOS)
+    if SO_PASSCRED is not None:
+        sock.setsockopt(SOL_SOCKET, SO_PASSCRED, 1)
+
     with close_on_error(sock):
         await sock.connect(os.fspath(filename))
 
     return trio.SocketStream(sock)
 
 
-def get_peer_info(sock: trio.socket.socket) -> tuple[
+def get_peer_pid(sock) -> int|None:
+    '''
+    Gets the PID of the process connected to the other end of a Unix
+    domain socket on macOS, or `None` if that fails.
+
+    NOTE, should work on MacOS (and others?).
+
+    '''
+    # try to get the peer PID using a naive soln found from,
+    # https://stackoverflow.com/a/67971484
+    #
+    # NOTE, a more correct soln is likely needed here according to
+    # the complaints of `copilot` which led to digging into the
+    # underlying `go`lang issue linked from the above SO answer,
+
+    # XXX, darwin-xnu kernel srces defining these constants,
+    # - SOL_LOCAL
+    # |_https://github.com/apple/darwin-xnu/blob/main/bsd/sys/un.h#L85
+    # - LOCAL_PEERPID
+    # |_https://github.com/apple/darwin-xnu/blob/main/bsd/sys/un.h#L89
+    #
+    SOL_LOCAL: int = 0
+    LOCAL_PEERPID: int = 0x002
+
+    try:
+        pid: int = sock.getsockopt(
+            SOL_LOCAL,
+            LOCAL_PEERPID,
+        )
+        return pid
+    except socket_error as e:
+        log.exception(
+            f"Failed to get peer PID: {e}"
+        )
+        return None
+
+
+def get_peer_info(
+    sock: trio.socket.socket,
+) -> tuple[
     int,  # pid
     int,  # uid
     int,  # guid
 ]:
     '''
     Deliver the connecting peer's "credentials"-info as defined in
-    a very Linux specific way..
+    a platform-specific way.
+
+    Linux-ONLY, uses SO_PEERCRED.
 
     For more deats see,
     - `man accept`,
@@ -337,6 +415,11 @@ def get_peer_info(sock: trio.socket.socket) -> tuple[
     - https://stackoverflow.com/a/7982749
 
     '''
+    if SO_PEERCRED is None:
+        raise RuntimeError(
+            f'Peer credential retrieval not supported on {sys.platform}!'
+        )
+
     creds: bytes = sock.getsockopt(
         SOL_SOCKET,
         SO_PEERCRED,
@@ -440,13 +523,37 @@ class MsgpackUDSStream(MsgpackTransport):
         match (peername, sockname):
             case (str(), bytes()):
                 sock_path: Path = Path(peername)
+
             case (bytes(), str()):
                 sock_path: Path = Path(sockname)
-        (
-            peer_pid,
-            _,
-            _,
-        ) = get_peer_info(sock)
+
+            case (str(), str()):  # XXX, likely macOS
+                sock_path: Path = Path(peername)
+
+            case _:
+                raise TypeError(
+                    f'Failed to match (peername, sockname) types?\n'
+                    f'peername: {peername!r}\n'
+                    f'sockname: {sockname!r}\n'
+                )
+
+        if sys.platform == 'linux':
+            (
+                peer_pid,
+                _,
+                _,
+            ) = get_peer_info(sock)
+
+        # NOTE known to at least works on,
+        # - macos
+        else:
+            peer_pid: int|None = get_peer_pid(sock)
+            if peer_pid is None:
+                log.warning(
+                    f'Unable to get peer PID?\n'
+                    f'sock: {sock!r}\n'
+                    f'peer_pid: {peer_pid!r}\n'
+                )
 
         filedir, filename = unwrap_sockpath(sock_path)
         laddr = UDSAddress(

tractor/log.py

@@ -293,7 +293,7 @@ _conc_name_getters = {
     'task': pformat_task_uid,
     'actor': lambda: _curr_actor_no_exc(),
     'actor_name': lambda: current_actor().name,
-    'actor_uid': lambda: current_actor().uid[1][:6],
+    'actor_uid': lambda: current_actor().aid.uuid[:6],
 }
 
 

tractor/msg/types.py

@@ -324,6 +324,8 @@ class Start(
     # => SEE ABOVE <=
     kwargs: dict[str, Any]
     uid: tuple[str, str]  # (calling) actor-id
+    # aid: Aid
+    # ^TODO, convert stat!
 
     # TODO: enforcing a msg-spec in terms `Msg.pld`
     # parameterizable msgs to be used in the appls IPC dialog.

tractor/to_asyncio.py

@@ -272,18 +272,16 @@ class LinkedTaskChannel(
         '''
         Receive a value `asyncio.Task` <- `trio.Task`.
 
-        This is equiv to `await self._from_trio.get()`.
+        This is equiv to `await self._to_aio.get()`.
 
         '''
         return await self._to_aio.get()
 
     async def send(self, item: Any) -> None:
         '''
-        Send a value through `trio.Task` -> `asyncio.Task`
-        presuming
-        it defines a `from_trio` argument or makes calls
-        to `chan.get()` , if it does not
-        this method will raise an error.
+        Send a value `trio.Task` -> `asyncio.Task`
+        by enqueuing `item` onto the internal
+        `asyncio.Queue` via `put_nowait()`.
 
         '''
         self._to_aio.put_nowait(item)
@@ -1304,13 +1302,14 @@ async def open_channel_from(
     tuple[LinkedTaskChannel, Any]
 ]:
     '''
-    Start an `asyncio.Task` as `target()` and open an inter-loop
-    (linked) channel for streaming between it and the current
-    `trio.Task`.
+    Start an `asyncio.Task` as `target()` and open an
+    inter-loop (linked) channel for streaming between
+    it and the current `trio.Task`.
 
-    A pair `(chan: LinkedTaskChannel, Any)` is delivered to the caller
-    where the 2nd element is the value provided by the
-    `asyncio.Task`'s unblocking call to `chan.started_nowait()`.
+    A pair `(chan: LinkedTaskChannel, Any)` is delivered
+    to the caller where the 2nd element is the value
+    provided by the `asyncio.Task`'s unblocking call
+    to `chan.started_nowait()`.
 
     '''
     chan: LinkedTaskChannel = _run_asyncio_task(
@@ -1334,8 +1333,7 @@ async def open_channel_from(
                     first = await chan.receive()
 
                     # deliver stream handle upward
-                    yield first, chan
-                    # ^TODO! swap these!!
+                    yield chan, first
             except trio.Cancelled as taskc:
                 if cs.cancel_called:
                     if isinstance(chan._trio_to_raise, AsyncioCancelled):

uv.lock

@@ -1,6 +1,6 @@
 version = 1
 revision = 3
-requires-python = ">=3.12"
+requires-python = ">=3.12, <3.14"
 
 [[package]]
 name = "async-generator"