Retract the hand-rolled tunnel peeler from plan-03

§3.2 specced a pure fn `_peel_tunnel_segs(proto_names) ->
(bearer_names, tunnel_specs, overlay_names)` to split a maddr at
its tunnel seg. It should never be written: `py-multiaddr` ships
that whole surface already and the plan simply missed it, even
though gh #443's 2nd bullet links the README sections in
question.

Replaced w/ a ⚠️ CORRECTION carrying the verified API table
(`.decapsulate_code(P_WG)` for the bearer, `.split()`/`.join()`
for a seg tail, `.value_for_protocol()` to read a value,
`.encapsulate()` to recompose) plus *why* it works on an infix
`/wg/` seg: the cut is by proto-code, never by matching an addr
value, and the key seg has no addr of its own.

Also,
- adopt `bearer`/`overlay` as the role names throughout, and say
  plainly why not `inner`/`outer` — the call-stack reading of
  "inner" is the exact opposite of the encapsulation one.
- warn that `value_for_protocol('ip4')` on a full tunnelled
  maddr silently yields the *bearer's* host; only call it on a
  peeled sub-maddr.
- note nesting (wg-in-wg) falls out of `.decapsulate_code()`
  cutting at the *last* occurrence, so peel repeatedly rather
  than recursing through a bespoke splitter.
- `mk_maddr()` for `TunnelledAddress` is `.encapsulate()`
  composition, not `str` building.
- README: drop the "degrades to a plain segment split" para,
  since that path is gone — no codec now means one actionable
  raise.

(this patch was generated in some part by `claude-code` using `claude-opus-5` (`anthropic`))

ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.md

@@ -1,62 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-session: pr512-docs-review
-timestamp: 2026-08-28T20:08:22Z
-git_ref: wkt/big_boi_docs_472_follow_ups
-scope: code
-substantive: true
-raw_file: 20260828T200822Z_0be872ff_prompt_io.raw.md
----
-
-## Prompt
-
-Prime PR #512 for landing by applying the repository Python style and
-typing policy across changed examples, reviewing and correcting the
-result with delegated agents, validating it, and auditing documentation
-changes since PR #460 for broader follow-up work.
-
-## Response summary
-
-Completed a branch-wide style, typing, and docstring pass over the
-changed examples. Reworked the dedicated registrar example so it proves
-external discovery, handles bind collisions, publishes readiness
-atomically, and shuts down cleanly across platforms. Staged review moved
-it under the discovery examples, introduced an async lifecycle manager,
-and captured public-subsystem, Piker, and pytest follow-ups. Corrected
-discovery and typed-messaging documentation after cross-checking runtime
-behavior.
-Created issue #514 for typed `Start` contracts and linked its planned
-sender-side argument validation from the messaging guide.
-Independent review passes were repeated until no actionable findings
-remained. Static checks, Sphinx, runnable examples, Trio coverage, and
-focused registrar checks passed; Ctrl-C debugger failures remain
-unclassified after one case reproduced on untouched `main`.
-
-## Files changed
-
-- `examples/discovery/dedicated_registrar.py` - implement robust
-  standalone registrar discovery and teardown.
-- `examples/**/*.py` - apply style, typing, docstring, and terminology
-  corrections to the 38 changed example modules.
-- `docs/guide/discovery.rst` - document actual registrar and
-  multi-registrar behavior.
-- `docs/guide/msging.rst` - document actual validation and codec-hook
-  boundaries.
-
-## Human edits
-
-The human selected PR #512, required the `/py-codestyle` policy and
-typing improvements throughout the examples, required delegated review
-agents to find additional fixes, authorized agents to own and push their
-code through the proper review process, and requested a broad audit of
-documentation changes since PR #460 rendered as an evolving Taken WIP.
-The human then required the earlier `ParamSpec` proposal to be captured
-as a dedicated GitHub issue and linked from the current docs patch.
-The human also required the registrar demo to move under discovery, use
-structural matching and an async context manager, explain socket ownership,
-and preserve public API, Piker, and pytest-harness follow-ups.
-The human also explicitly retracted messages intended for another
-session; those messages did not direct the result. No direct manual
-source edits were identified; the agent applied the human-directed
-changes.

ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.raw.md

@@ -1,90 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-08-28T20:08:22Z
-git_ref: wkt/big_boi_docs_472_follow_ups
-diff_cmd: git diff 0be872ff..HEAD
----
-
-## Prompt
-
-Prime PR #512 for landing. Apply the Python style policy throughout
-the changed examples, improve typing, use review agents to find and
-fix correctness issues, run the proper review process, push the agent's
-own code, and then audit documentation changes since PR #460 for stale,
-missing, defunct, or improvable coverage.
-
-## Response
-
-> `git diff 0be872ff..HEAD -- examples/`
-
-Restyled and typed all 38 changed example modules while preserving
-intentional debugger failures and hanging reproducers. Added required
-multiline docstrings, corrected async callable annotations, modernized
-legacy API terminology, and preserved complete debugger process trees.
-
-Rebuilt `examples/discovery/dedicated_registrar.py` as a real
-external-discovery example. A standalone registrar process now selects
-and retries bounded ephemeral addresses, publishes readiness atomically,
-rejects accidental registrar reuse, serves sibling service and client
-actors, proves lookup used the registrar instead of a local-peer channel,
-and performs bounded, validated, cross-platform shutdown.
-
-Staged review moved the demo into the discovery example group, converted
-collision classification to structural pattern matching, documented the
-selector-socket close/rebind race, and extracted process ownership into an
-async context manager. A source TODO records the future public discovery
-subsystem, Piker service-management lessons, and pytest registry-isolation
-use case.
-
-> `git diff 0be872ff..HEAD -- docs/guide/discovery.rst`
-
-Corrected registrar, duplicate-name, and multi-registrar discovery
-guidance, including configured-order and `None` placeholder behavior.
-
-> `git diff 0be872ff..HEAD -- docs/guide/msging.rst`
-
-Corrected typed-payload validation boundaries and separated working
-task-scoped codec encoding from private per-dialog decoding and the
-incomplete decorator hook parameters.
-
-## Review and validation
-
-Multiple independent reviews found and drove fixes for registrar
-discovery validity, port-selection races, teardown, process diagnostics,
-shutdown status, inaccurate discovery ordering, async callable typing,
-missing docstrings, truncated debugger command diagrams, stale APIs, and
-payload-error relay wording.
-
-Validation completed:
-
-- AST parsing, Ruff, 69-column checks, and required-docstring audit for
-  all 38 changed Python files.
-- Sphinx HTML build succeeded.
-- Documentation example harness: 24 passed.
-- Trio coverage: 7 passed, 4 xfailed, 1 xpassed.
-- Dedicated registrar direct run and focused harness test passed with a
-  clean child exit and no traceback.
-- Debugger suite: 20 passed, 6 skipped, and 8 reproducible
-  `ctl-c=True` pexpect timeouts. One exact failure reproduced on
-  untouched `main`, confirming a baseline failure; its root cause
-  remains unclassified. No leaked actor processes remained.
-
-The post-PR-#460 audit identified follow-up work around advertised but
-inert runtime selectors, platform/backend support, unresolved discovery
-contracts, cached-context teardown, codec recipes, examples-as-tests
-coverage, public API exports, broadcast factory contracts, stale examples,
-README duplication, release notes, and process-title terminology.
-
-## Follow-up prompt
-
-Capture the previously proposed `ParamSpec`-based `Start` argument
-validation work in a focused GitHub issue, then link it from the current
-typed-messaging docs patch as planned sender-side checking.
-
-## Follow-up response
-
-Created https://github.com/goodboy/tractor/issues/514 to track deriving
-typed `Start` contracts from endpoint signatures, preserving caller-facing
-signatures, and validating arguments before sending where possible. Added
-the issue link beside the guide's current `Start` validation boundary.

docs/guide/discovery.rst

@@ -30,9 +30,9 @@ the registry tracks the live tree as it grows and shrinks.
 
 .. note::
    Actor names are **not** enforced unique — the registry is keyed
-   by the full ``(name, uuid)`` pair. A name lookup returns one
+   by the full ``(name, uuid)`` pair. Name-based lookups simply
-   matching registration, but the API does not promise which match
+   resolve to the *last* registered match, so if you boot five
-   wins. Use unique service names when selection matters.
+   actors all named ``'bob'``, you get the freshest ``'bob'`` B)
 
 First boot: who's the registrar?
 --------------------------------
@@ -65,43 +65,6 @@ the one-and-only registrar; boot then fails loudly with a
 ``RuntimeError`` if some other process already bound the registry
 socket(s).
 
-A dedicated registrar
----------------------
-That second rule — *"if a registrar answers, boot as a plain
-root"* — is all you need to run the registry as its own
-**standalone process**, decoupled from any app tree's root. In the
-daemon process, enter ``open_root_actor()`` with an explicit
-``registry_addrs`` and ``ensure_registry=True``; the latter makes
-startup fail instead of silently joining a registrar that won the
-address. Point each app tree at the address that daemon actually
-bound:
-
-.. literalinclude:: ../../examples/discovery/dedicated_registrar.py
-   :caption: examples/discovery/dedicated_registrar.py
-   :language: python
-
-The example's selector socket binds but deliberately never listens.
-It owns the kernel-selected local address only long enough to read it,
-then closes so Tractor's actual listener can bind the same address.
-This is not a socket transfer: the close/rebind handoff is non-atomic,
-so the example retries with a fresh candidate only when registrar
-startup reports that another process claimed the released address.
-Retries are bounded, and other startup failures remain visible. It
-publishes the selected address only after the actor context enters.
-It also performs the lookup inside a separate ``client`` actor. The
-service is its sibling, not its child, so the client has no spawn-time
-service channel to satisfy the local-peer fast path. The
-``query_actor()`` assertion verifies that a registrar portal handled
-the lookup before ``find_actor()`` makes the service RPC.
-
-This is the "registrar as a subsystem, not the app root actor"
-shape. Two caveats today (both tracked as #472 follow-ups):
-``enable_transports`` is single-proto per runtime, so a registrar
-can't yet serve multiple backends at once; and there's no way to
-spawn a registrar as a *sub*-actor of a shared tree (only as its
-own root), since ``start_actor()`` has no custom-``actor_cls``
-hook.
-
 Looking up actors
 -----------------
 
@@ -129,22 +92,13 @@ Knobs worth knowing:
 - ``registry_addrs=[...]``: query specific (possibly multiple,
   possibly remote) registrars instead of your tree's default,
 
-- ``only_first=True``: after all configured registrars are queried
+- ``only_first=False``: deliver a ``list[Portal]`` of *all*
-  concurrently, yield the result in the first ``registry_addrs``
+  matches found across the queried registrars instead of just the
-  position. This is configured order, not first-reachable order, so
+  first,
-  the result can be ``None`` even when a later registrar returned a
-  portal,
 
-- ``only_first=False``: when any query succeeds, yield an ordered
+- ``raise_on_none=True``: raise a ``RuntimeError`` instead of
-  ``list[Portal | None]`` with one result per ``registry_addrs``
+  yielding ``None`` when no match is found — for when absence is
-  position; misses remain ``None`` placeholders. When every query
+  a hard error in your app.
-  misses, yield ``None`` instead of a list. This does not enumerate
-  every duplicate name in one registrar,
-
-- ``raise_on_none=True``: raise a ``RuntimeError`` when every
-  registrar query returns ``None``. With ``only_first=True`` it does
-  not raise merely because the first ordered result is ``None`` when
-  a later result is a portal.
 
 ``wait_for_actor()``
 ********************
@@ -177,11 +131,10 @@ Yields a portal straight to the registrar actor itself — or a
 Fast paths and address preference
 ---------------------------------
 
-Before doing any RPC to the registrar, ``query_actor()``,
+Before doing any RPC to the registrar, every lookup first scans
-``wait_for_actor()``, and the default ``find_actor()`` lookup first
+the calling actor's *already-connected peers*: if you have a live
-scan the calling actor's *already-connected peers*. If the caller
+channel to an actor named ``name`` you get a portal over it
-has a live channel to an actor named ``name``, it gets a portal over
+immediately, no registrar round-trip at all.
-that channel immediately, with no registrar round-trip.
 
 When a registry entry holds *multiple* addresses (a multihomed
 actor) the "best" one is chosen by locality:
@@ -248,8 +201,8 @@ the existing registrar:
 
 Per the bootstrap rules above, if those addrs are absent this process
 becomes its own registrar root, so the same code works standalone and
-as a tree-joiner. An occupied address that does not complete a
+as a tree-joiner. An occupied address that does not complete a Tractor
-Tractor registrar handshake fails startup instead of being rebound.
+registrar handshake fails startup instead of being rebound.
 
 "Arbiter"? A legacy naming note
 -------------------------------
@@ -273,11 +226,10 @@ Very naive, very honest
 -----------------------
 
 To be clear, this is a **very naive** discovery system: one
-in-memory registrar holding a dict, no replication, no re-election
+process-tree-local registrar holding a dict, no replication, no
-when it dies, and no automatic cross-host propagation. Separate
+re-election when it dies, no cross-host propagation. That's
-programs can use the same reachable registrar, as above, but must be
+intentional (for now); it covers the "wire up my services on this
-configured with its address. That's intentional (for now); it covers
+host" case without dragging in a consensus protocol.
-the "wire up my services" case without a consensus protocol.
 
 On the roadmap (issue `#216`_ tracks a chunk of it):
 
@@ -302,7 +254,6 @@ to hear from you.
      :class:`tractor.Registrar`.
 
 .. _gossip protocol: https://en.wikipedia.org/wiki/Gossip_protocol
-.. _modern protocol:
+.. _modern protocol: https://en.wikipedia.org/wiki/Rendezvous_protocol
-   https://en.wikipedia.org/wiki/Rendezvous_protocol
 .. _discovery: https://zguide.zeromq.org/docs/chapter8/#Discovery
 .. _#216: https://github.com/goodboy/tractor/issues/216

docs/guide/msging.rst

@@ -153,16 +153,14 @@ the high-rate stream path.
   never even hits the wire. (You can opt out per-call with
   ``ctx.started(..., validate_pld_spec=False)`` if you measure
   a real cost.)
-- ``Yield`` and ``Return`` payloads are not checked before sending;
+- ``Yield`` payloads are **never** checked inside
-  they're decoded against the dialog's spec by the receiver. A
+  ``MsgStream.send()``; they're validated receiver-side on each
-  violation raises a ``MsgTypeError`` there and terminates that
+  ``MsgStream.receive()``. A violation raises a ``MsgTypeError``
-  dialog. The peer then observes the resulting protocol teardown;
+  in the receiver *and* relays an ``Error`` msg back so the
-  it is not guaranteed to receive the same ``MsgTypeError``.
+  offending sender gets one raised too.
-- ``Start`` arguments are dispatched through the RPC endpoint's
+- the remaining control msgs (``Start``, ``Return``) are likewise
-  Python signature. They are not payloads covered by the dialog's
+  validated such that violations raise in the **sending** actor,
-  ``pld_spec``. A planned follow-up will derive a typed ``Start``
+  pointing the traceback at the code that actually goofed.
-  contract from endpoint annotations and validate arguments
-  sender-side; see `#514`_.
 
 Anatomy of a ``MsgTypeError``
 -----------------------------
@@ -179,21 +177,21 @@ a msg fails to decode against the active spec. The useful bits:
   ``.src_uid``, ``.ipc_msg`` and the fancy ``.pformat()`` tb-box
   rendering.
 
-Practical reading guide: a *sender-side* MTE for ``Started`` points
+Practical reading guide: a *sender-side* MTE (``Started``,
-straight at the offending ``await ctx.started()`` call. A
+``Return``) points straight at your offending ``await
-*receiver-side* MTE for ``Yield`` or ``Return`` surfaces while the
+ctx.started()`` or ``return`` statement, while a *receiver-side*
-peer decodes the payload. Either way the failure is scoped to that
+MTE (``Yield``) surfaces from the consumer's ``receive()`` call
-one dialog; sibling contexts on the same channel keep right on
+with the relay copy delivered back to the producer. Either way
-trucking.
+the failure is scoped to that one dialog; sibling contexts on the
+same channel keep right on trucking.
 
 Custom wire types: ``mk_codec()`` and friends
 ---------------------------------------------
 msgspec covers a wide set of `builtin types`__ natively; for
-anything else you teach the codec via extension hooks. The complete
+anything else you teach the codec via extension hooks. The
-public path currently available is task-scoped encoding:
+easiest path is per-endpoint: ``@tractor.context()`` accepts
-``tractor.msg.mk_codec()`` builds a codec with an ``enc_hook``, and
+``enc_hook``/``dec_hook`` params right alongside ``pld_spec``.
-``tractor.msg.apply_codec()`` installs it for the current task. To
+For full control build and apply a codec yourself; encode-side:
-build and apply that transport codec:
 
 __ https://jcristharif.com/msgspec/supported-types.html
 
@@ -208,9 +206,8 @@ __ https://jcristharif.com/msgspec/supported-types.html
     with apply_codec(codec):        # ContextVar-scoped override
         ...  # msgs sent by this task now encode NSPs
 
-The context manager which temporarily installs payload-decoder
+and decode-side, scoped to an open context (note the import from
-settings on an open context is separate and still private (note
+``tractor.msg._ops``, not yet re-exported):
-the ``tractor.msg._ops`` import):
 
 .. code:: python
 
@@ -223,15 +220,11 @@ the ``tractor.msg._ops`` import):
     ):
         ...  # this dialog's payloads decode as NSPs
 
-``apply_codec()`` is ``ContextVar``-scoped: it overrides the codec
+``apply_codec()`` is ``ContextVar``-scoped: it overrides the
-for the current task (and only that task), not the whole process.
+codec for the current task (and only that task), not the whole
-``@tractor.context()`` accepts ``enc_hook`` and ``dec_hook``
+process. For complete working flows, including hook pairing rules
-parameters, but their runtime wiring is not yet a symmetric,
+and roundtrip cases, see ``tests/msg/test_ext_types_msgspec.py``
-end-to-end public hook pair: the encode hook is not consumed and
+and ``tests/msg/test_pldrx_limiting.py``.
-the decode hook is not applied on both peers. For the working flows
-and their current boundaries, see
-``tests/msg/test_ext_types_msgspec.py`` and
-``tests/msg/test_pldrx_limiting.py``.
 
 The runtime dogfoods this pattern with
 :class:`tractor.msg.NamespacePath`: a ``str``-subtype shaped like
@@ -245,33 +238,9 @@ Toward capability-based msging
 The ``pld_spec`` + codec-hook layer is the foundation for the
 long-game: **capability-based msging** where each dialog's
 type contract doubles as a capability grant, negotiated as part
-of the protocol itself. The epic is tracked in `#196`_ (evolving
+of the protocol itself. That work is tracked in `#196`_ (with the
-the original typed-proto work in `#36`_), and the most recent
+original typed-proto epic in `#36`_); if strongly-typed
-concrete step is `#365`_ — driving the whole ``pld_spec`` off
+distributed systems get you going, we'd love your input.
-plain type-annotations (e.g. annotating a context's
-``open_stream()`` with ``msgspec.Struct`` subtypes) instead of
-explicit ``pld_spec=`` kwargs.
-
-You don't have to wait for that, though: the decorator-level
-``@tractor.context(pld_spec=...)`` shown above is already the
-*higher-level* way to pin a dialog's payload contract, while
-``tractor.msg._ops.limit_plds()`` is the lower-level, per-block
-escape hatch. Both are exercised end-to-end in
-``tests/msg/test_pldrx_limiting.py`` and
-``tests/msg/test_ext_types_msgspec.py``.
-
-The codec constructor and task-scoped override are public; the
-per-dialog decoder override remains private, and the decorator hook
-parameters remain incomplete. `#376`_ (from
-`@guilledk <https://github.com/guilledk>`_, on the
-`auto_codecs <https://github.com/goodboy/tractor/tree/auto_codecs>`_
-branch) instead drafts pair-building factories which derive
-matching ``enc_hook``/``dec_hook`` functions and encoder/decoder
-pairs from a type spec. That automation, not hook availability,
-is the proposed long-term home for custom-type (de)serialization.
-
-If strongly-typed distributed systems get you going, we'd love
-your input on any of the above.
 
 Where to next?
 --------------
@@ -289,6 +258,3 @@ Where to next?
 .. _(un)protocol: https://zguide.zeromq.org/docs/chapter7/#Unprotocols
 .. _#196: https://github.com/goodboy/tractor/issues/196
 .. _#36: https://github.com/goodboy/tractor/issues/36
-.. _#365: https://github.com/goodboy/tractor/issues/365
-.. _#376: https://github.com/goodboy/tractor/pull/376
-.. _#514: https://github.com/goodboy/tractor/issues/514

examples/a_trynamic_first_scene.py

@@ -1,40 +1,28 @@
 import trio
 import tractor
 
-_this_module: str = __name__
+_this_module = __name__
-the_line: str = 'Hi my name is {}'
+the_line = 'Hi my name is {}'
 
 
-tractor.log.get_console_log('INFO')
+tractor.log.get_console_log("INFO")
 
 
-async def hi() -> str:
+async def hi():
-    '''
-    Return a greeting naming the current actor.
-
-    '''
     return the_line.format(tractor.current_actor().name)
 
 
-async def say_hello(other_actor: str) -> str:
+async def say_hello(other_actor):
-    '''
-    Ask another actor to return its greeting.
-
-    '''
-    portal: tractor.Portal
     async with tractor.wait_for_actor(other_actor) as portal:
         return await portal.run(hi)
 
 
-async def main() -> None:
+async def main():
-    '''
+    """Main tractor entry point, the "master" process (for now
-    Main tractor entry point, the "master" process (for now
     acts as the "director").
-
+    """
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        print('Alright... Action!')
+        print("Alright... Action!")
 
         # both actors wait on (then dial!) the *other*, so each
         # must outlive both hellos: spawn as daemons, run the
@@ -51,10 +39,6 @@ async def main() -> None:
             name: str,
             other_actor: str,
         ) -> None:
-            '''
-            Print a greeting fetched through a named actor.
-
-            '''
             print(
                 # RPC through an existing actor's `Portal`.
                 await portals[name].run(
@@ -63,14 +47,13 @@ async def main() -> None:
                 )
             )
 
-        tn: trio.Nursery
         async with trio.open_nursery() as tn:
             tn.start_soon(run_and_print, 'donny', 'gretchen')
             tn.start_soon(run_and_print, 'gretchen', 'donny')
 
         await an.cancel()
 
-    print('CUTTTT CUUTT CUT!!! Donny!! You\'re supposed to say...')
+    print("CUTTTT CUUTT CUT!!! Donny!! You're supposed to say...")
 
 
 if __name__ == '__main__':

examples/actor_spawning_and_causality.py

@@ -2,20 +2,14 @@ import trio
 import tractor
 
 
-async def cellar_door() -> str:
+async def cellar_door():
-    '''
-    Return a phrase from a spawned actor.
-
-    '''
     assert not tractor.is_root_process()
-    return 'Dang that\'s beautiful'
+    return "Dang that's beautiful"
 
 
-async def main() -> None:
+async def main():
-    '''
+    """The main ``tractor`` routine.
-    The main ``tractor`` routine.
+    """
-
-    '''
     # spawn a subactor, run ``cellar_door()`` as its lone task,
     # block until its result arrives and the subactor is reaped.
     print(

examples/actor_spawning_and_causality_with_daemon.py

@@ -2,24 +2,19 @@ import trio
 import tractor
 
 
-async def movie_theatre_question() -> str:
+async def movie_theatre_question():
-    '''
+    """A question asked in a dark theatre, in a tangent
-    A question asked in a dark theatre, in a tangent
     (errr, I mean different) process.
-
+    """
-    '''
     return 'have you ever seen a portal?'
 
 
-async def main() -> None:
+async def main():
-    '''
+    """The main ``tractor`` routine.
-    The main ``tractor`` routine.
+    """
-
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
 
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'frank',
             # enable the actor to run funcs from this current module
             enable_modules=[__name__],
@@ -29,8 +24,9 @@ async def main() -> None:
         # call the subactor a 2nd time
         print(await portal.run(movie_theatre_question))
 
-        # the async with will wait indefinitely for "frank" because
+        # the async with will block here indefinitely waiting
-        # its runtime remains active until explicitly cancelled
+        # for our actor "frank" to complete, but since it's an
+        # "outlive_main" actor it will never end until cancelled
         await portal.cancel_actor()
 
 

examples/asynchronous_generators.py

@@ -1,46 +1,33 @@
-from itertools import repeat
 from typing import AsyncIterator
+from itertools import repeat
 
 import trio
 import tractor
 
 
-async def stream_forever() -> AsyncIterator[str]:
+async def stream_forever() -> AsyncIterator[int]:
-    '''
-    Stream the same message indefinitely.
 
-    '''
+    for i in repeat("I can see these little future bubble things"):
-    message: str
+        # each yielded value is sent over the ``Channel`` to the parent actor
-    for message in repeat(
+        yield i
-        'I can see these little future bubble things',
-    ):
-        # each yielded value is sent over the ``Channel`` to the
-        # parent actor
-        yield message
         await trio.sleep(0.01)
 
 
-async def main() -> None:
+async def main():
-    '''
-    Print messages streamed from a subactor.
 
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
 
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'donny',
             enable_modules=[__name__],
         )
 
         # this async for loop streams values from the above
         # async generator running in a separate process
-        stream: tractor.MsgStream
         async with portal.open_stream_from(stream_forever) as stream:
-            count: int = 0
+            count = 0
-            message: str
+            async for letter in stream:
-            async for message in stream:
+                print(letter)
-                print(message)
                 count += 1
 
                 if count > 50:

examples/debugging/fast_error_in_root_after_spawn.py

@@ -13,11 +13,7 @@ import tractor
 @tractor.context
 async def sleep(
     ctx: tractor.Context,
-) -> None:
+):
-    '''
-    Start a context after a brief initialization delay.
-
-    '''
     await trio.sleep(0.5)
     await ctx.started()
     await trio.sleep_forever()
@@ -25,13 +21,10 @@ async def sleep(
 
 async def open_ctx(
     n: tractor.runtime._supervise.ActorNursery
-) -> None:
+):
-    '''
-    Spawn a sleeper and open a context with it.
 
-    '''
     # spawn both actors
-    portal: tractor.Portal = await n.start_actor(
+    portal = await n.start_actor(
         name='sleeper',
         enable_modules=[__name__],
     )
@@ -42,11 +35,8 @@ async def open_ctx(
         assert first is None
 
 
-async def main() -> None:
+async def main():
-    '''
-    Fail the root while a subactor context is still starting.
 
-    '''
     async with tractor.open_nursery(
         debug_mode=True,
         loglevel='runtime',

examples/debugging/multi_daemon_subactors.py

@@ -1,14 +1,9 @@
-from collections.abc import AsyncIterator
-
 import tractor
 import trio
 
 
-async def breakpoint_forever() -> AsyncIterator[str]:
+async def breakpoint_forever():
-    '''
+    "Indefinitely re-enter debugger in child actor."
-    Indefinitely re-enter debugger in child actor.
-
-    '''
     try:
         while True:
             yield 'yo'
@@ -20,15 +15,12 @@ async def breakpoint_forever() -> AsyncIterator[str]:
         raise
 
 
-async def name_error() -> None:
+async def name_error():
-    '''
+    "Raise a ``NameError``"
-    Raise a ``NameError``.
-
-    '''
     getattr(doggypants)  # noqa
 
 
-async def main() -> None:
+async def main():
     '''
     Test breakpoint in a streaming actor.
 
@@ -36,14 +28,8 @@ async def main() -> None:
     async with tractor.open_nursery(
         debug_mode=True,
     ) as an:
-        p0: tractor.Portal = await an.start_actor(
+        p0 = await an.start_actor('bp_forever', enable_modules=[__name__])
-            'bp_forever',
+        p1 = await an.start_actor('name_error', enable_modules=[__name__])
-            enable_modules=[__name__],
-        )
-        p1: tractor.Portal = await an.start_actor(
-            'name_error',
-            enable_modules=[__name__],
-        )
 
         # retreive results
         async with p0.open_stream_from(breakpoint_forever) as stream:
@@ -54,7 +40,6 @@ async def main() -> None:
             except tractor.RemoteActorError as rae:
                 assert rae.boxed_type is NameError
 
-            i: str
             async for i in stream:
 
                 # a second time try the failing subactor and this tie

examples/debugging/multi_nested_subactors_error_up_through_nurseries.py

@@ -4,19 +4,13 @@ import trio
 import tractor
 
 
-async def name_error() -> None:
+async def name_error():
-    '''
+    "Raise a ``NameError``"
-    Raise a ``NameError``.
-
-    '''
     getattr(doggypants)  # noqa
 
 
-async def breakpoint_forever() -> None:
+async def breakpoint_forever():
-    '''
+    "Indefinitely re-enter debugger in child actor."
-    Indefinitely re-enter debugger in child actor.
-
-    '''
     while True:
         await tractor.pause()
 
@@ -26,13 +20,9 @@ async def breakpoint_forever() -> None:
         # await trio.sleep(0)
 
 
-async def spawn_until(
+async def spawn_until(depth=0):
-    depth: int = 0,
+    """"A nested nursery that triggers another ``NameError``.
-) -> None:
+    """
-    '''
-    A nested nursery that triggers another ``NameError``.
-
-    '''
     async with (
         tractor.open_nursery() as an,
         trio.open_nursery() as tn,
@@ -47,8 +37,7 @@ async def spawn_until(
                 )
             )
 
-            # Let the background one-shot enter
+            # Let the background one-shot enter `breakpoint_forever()`
-            # `breakpoint_forever()`
             # before its sibling raises and cancellation propagates.
             await trio.sleep(0.5)
             # rx and propagate error from child
@@ -59,9 +48,9 @@ async def spawn_until(
             )
 
         else:
-            # recusrive call to spawn another process branching
+            # recusrive call to spawn another process branching layer of
-            # layer of the tree; blocks (up) each level until the
+            # the tree; blocks (up) each level until the leaf's
-            # leaf's `name_error` relays through.
+            # `name_error` relays through.
             depth -= 1
             await tractor.to_actor.run(
                 partial(
@@ -74,35 +63,25 @@ async def spawn_until(
 
 
 # TODO: notes on the new boxed-relayed errors through proxy actors
-async def main() -> None:
+async def main():
-    '''
+    """The main ``tractor`` routine.
-    The main ``tractor`` routine.
 
-    The process tree should look approximately as follows when the
+    The process tree should look as approximately as follows when the debugger
-    debugger first engages:
+    first engages:
 
     python examples/debugging/multi_nested_subactors_bp_forever.py
-    ├─ python -m tractor._child --uid
+    ├─ python -m tractor._child --uid ('spawner1', '7eab8462 ...)
-    │  ('spawner1', '7eab8462 ...')
+    │  └─ python -m tractor._child --uid ('spawn_until_3', 'afcba7a8 ...)
-    │  └─ python -m tractor._child --uid
+    │     └─ python -m tractor._child --uid ('spawn_until_2', 'd2433d13 ...)
-    │     ('spawn_until_3', 'afcba7a8 ...')
+    │        └─ python -m tractor._child --uid ('spawn_until_1', '1df589de ...)
-    │     └─ python -m tractor._child --uid
+    │           └─ python -m tractor._child --uid ('spawn_until_0', '3720602b ...)
-    │        ('spawn_until_2', 'd2433d13 ...')
-    │        └─ python -m tractor._child --uid
-    │           ('spawn_until_1', '1df589de ...')
-    │           └─ python -m tractor._child --uid
-    │              ('spawn_until_0', '3720602b ...')
     │
-    └─ python -m tractor._child --uid
+    └─ python -m tractor._child --uid ('spawner0', '1d42012b ...)
-       ('spawner0', '1d42012b ...')
+       └─ python -m tractor._child --uid ('spawn_until_2', '2877e155 ...)
-       └─ python -m tractor._child --uid
+          └─ python -m tractor._child --uid ('spawn_until_1', '0502d786 ...)
-          ('spawn_until_2', '2877e155 ...')
+             └─ python -m tractor._child --uid ('spawn_until_0', 'de918e6d ...)
-          └─ python -m tractor._child --uid
-             ('spawn_until_1', '0502d786 ...')
-             └─ python -m tractor._child --uid
-                ('spawn_until_0', 'de918e6d ...')
 
-    '''
+    """
     async with (
         tractor.open_nursery(
             debug_mode=True,

examples/debugging/multi_subactor_root_errors.py

@@ -7,19 +7,14 @@ import trio
 import tractor
 
 
-async def name_error() -> None:
+async def name_error():
-    '''
+    "Raise a ``NameError``"
-    Raise a ``NameError``.
-
-    '''
     getattr(doggypants)  # noqa
 
 
-async def spawn_error() -> None:
+async def spawn_error():
-    '''
+    """"A nested nursery that triggers another ``NameError``.
-    A nested nursery that triggers another ``NameError``.
+    """
-
-    '''
     async with tractor.open_nursery() as an:
         return await tractor.to_actor.run(
             name_error,
@@ -28,9 +23,8 @@ async def spawn_error() -> None:
         )
 
 
-async def main() -> None:
+async def main():
-    '''
+    """The main ``tractor`` routine.
-    The main ``tractor`` routine.
 
     The process tree should look as approximately as follows:
 
@@ -43,8 +37,7 @@ async def main() -> None:
         - nested name_error sub-sub-actor
         - root actor should then fail on assert
         - program termination
-
+    """
-    '''
     async with (
         tractor.open_nursery(
             debug_mode=True,
@@ -53,11 +46,11 @@ async def main() -> None:
         trio.open_nursery() as tn,
     ):
         # spawn both actors..
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'name_error',
             enable_modules=[__name__],
         )
-        portal1: tractor.Portal = await an.start_actor(
+        portal1 = await an.start_actor(
             'spawn_error',
             enable_modules=[__name__],
         )

examples/debugging/multi_subactors.py

@@ -1,32 +1,22 @@
-from collections.abc import Awaitable, Callable
-
 import tractor
 import trio
 
 
-async def breakpoint_forever() -> None:
+async def breakpoint_forever():
-    '''
+    "Indefinitely re-enter debugger in child actor."
-    Indefinitely re-enter debugger in child actor.
-
-    '''
     while True:
         await trio.sleep(0.1)
         await tractor.pause()
 
 
-async def name_error() -> None:
+async def name_error():
-    '''
+    "Raise a ``NameError``"
-    Raise a ``NameError``.
-
-    '''
     getattr(doggypants)  # noqa
 
 
-async def spawn_error() -> None:
+async def spawn_error():
-    '''
+    """"A nested nursery that triggers another ``NameError``.
-    A nested nursery that triggers another ``NameError``.
+    """
-
-    '''
     async with tractor.open_nursery() as an:
         return await tractor.to_actor.run(
             name_error,
@@ -35,9 +25,8 @@ async def spawn_error() -> None:
         )
 
 
-async def main() -> None:
+async def main():
-    '''
+    """The main ``tractor`` routine.
-    The main ``tractor`` routine.
 
     The process tree should look as approximately as follows:
 
@@ -46,8 +35,7 @@ async def main() -> None:
     |-python -m tractor._child --uid ('bp_forever', '1f787a7e ...)
     `-python -m tractor._child --uid ('spawn_error', '52ee14a5 ...)
        `-python -m tractor._child --uid ('name_error', '3391222c ...)
-
+    """
-    '''
     errors: list[BaseException] = []
 
     async with tractor.open_nursery(
@@ -55,14 +43,12 @@ async def main() -> None:
         # loglevel='runtime',
     ) as an:
 
-        async def run_and_collect(
+        async def run_and_collect(fn):
-            fn: Callable[[], Awaitable[object]],
-        ) -> None:
             '''
             One-shot whose (boxed) error is stashed instead of
             raised so a sibling's crash never cancels the others
             before they've had their own debugger sessions (the
-            'collect all errors' the legacy `run_in_actor()` API
+            "collect all errors" the legacy `run_in_actor()` API
             did implicitly at nursery teardown).
 
             '''

examples/debugging/per_actor_debug.py

@@ -1,28 +1,19 @@
 import trio
 import tractor
 
-
+async def die():
-async def die() -> None:
-    '''
-    Deliberately crash the calling actor.
-
-    '''
     raise RuntimeError
 
 
-async def main() -> None:
+async def main():
-    '''
-    Crash actors with different debugger settings concurrently.
-
-    '''
     async with tractor.open_nursery() as an:
 
-        debug_actor: tractor.Portal = await an.start_actor(
+        debug_actor = await an.start_actor(
             'debugged_boi',
             enable_modules=[__name__],
             debug_mode=True,
         )
-        crash_boi: tractor.Portal = await an.start_actor(
+        crash_boi = await an.start_actor(
             'crash_boi',
             enable_modules=[__name__],
             # debug_mode=True,

examples/debugging/pm_in_subactor.py

@@ -5,7 +5,7 @@ import tractor
 @tractor.context
 async def name_error(
     ctx: tractor.Context,
-) -> None:
+):
     '''
     Raise a `NameError`, catch it and enter `.post_mortem()`, then
     expect the `._rpc._invoke()` crash handler to also engage.
@@ -18,7 +18,7 @@ async def name_error(
         raise
 
 
-async def main() -> None:
+async def main():
     '''
     Test 3 `PdbREPL` entries:
       - one in the child due to manual `.post_mortem()`,
@@ -49,9 +49,7 @@ async def main() -> None:
             await tractor.post_mortem()
             raise
         else:
-            raise RuntimeError(
+            raise RuntimeError('IPC ctx should have remote errored!?')
-                'IPC ctx should have remote errored!?'
-            )
 
 
 if __name__ == '__main__':

examples/debugging/root_actor_breakpoint.py

@@ -2,11 +2,8 @@ import trio
 import tractor
 
 
-async def main() -> None:
+async def main():
-    '''
-    Pause in the root actor to exercise its debugger REPL.
 
-    '''
     async with tractor.open_root_actor(
         debug_mode=True,
     ):

examples/debugging/root_actor_error.py

@@ -2,11 +2,7 @@ import trio
 import tractor
 
 
-async def main() -> None:
+async def main():
-    '''
-    Raise an assertion error from the debug-enabled root actor.
-
-    '''
     async with tractor.open_root_actor(
         debug_mode=True,
     ):

examples/debugging/root_cancelled_but_child_is_in_tty_lock.py

@@ -4,21 +4,14 @@ import trio
 import tractor
 
 
-async def name_error() -> None:
+async def name_error():
-    '''
+    "Raise a ``NameError``"
-    Raise a ``NameError``.
-
-    '''
     getattr(doggypants)  # noqa
 
 
-async def spawn_until(
+async def spawn_until(depth=0):
-    depth: int = 0,
+    """"A nested nursery that triggers another ``NameError``.
-) -> None:
+    """
-    '''
-    A nested nursery that triggers another ``NameError``.
-
-    '''
     async with tractor.open_nursery() as an:
         if depth < 1:
             await tractor.to_actor.run(name_error, an=an)
@@ -34,23 +27,18 @@ async def spawn_until(
             )
 
 
-async def main() -> None:
+async def main():
     '''
     The process tree should look as approximately as follows when the
     debugger first engages:
 
     python examples/debugging/multi_nested_subactors_bp_forever.py
-    ├─ python -m tractor._child --uid
+    ├─ python -m tractor._child --uid ('spawner1', '7eab8462 ...)
-    │  ('spawner1', '7eab8462 ...')
+    │  └─ python -m tractor._child --uid ('spawn_until_0', '3720602b ...)
-    │  └─ python -m tractor._child --uid
+    │     └─ python -m tractor._child --uid ('name_error', '505bf71d ...)
-    │     ('spawn_until_0', '3720602b ...')
-    │     └─ python -m tractor._child --uid
-    │        ('name_error', '505bf71d ...')
     │
-    └─ python -m tractor._child --uid
+    └─ python -m tractor._child --uid ('spawner0', '1d42012b ...)
-       ('spawner0', '1d42012b ...')
+       └─ python -m tractor._child --uid ('name_error', '6c2733b8 ...)
-       └─ python -m tractor._child --uid
-          ('name_error', '6c2733b8 ...')
 
     '''
     async with (

examples/debugging/root_self_cancelled_w_error.py

@@ -2,11 +2,7 @@ import trio
 import tractor
 
 
-async def main() -> None:
+async def main():
-    '''
-    Enter shielded debugging after root cancellation, then fail.
-
-    '''
     async with tractor.open_root_actor(
         debug_mode=True,
         loglevel='cancel',
@@ -22,19 +18,16 @@ async def main() -> None:
         try:
             await tractor.pause()
         except trio.Cancelled as _taskc:
-            root_cs: trio.CancelScope
+            assert (root_cs := _root._root_tn.cancel_scope).cancel_called
-            assert (
-                root_cs := _root._root_tn.cancel_scope
-            ).cancel_called
             # NOTE^^ above logic but inside `open_root_actor()` and
             # passed to the `shield=` expression is effectively what
             # we're testing here!
             await tractor.pause(shield=root_cs.cancel_called)
 
-        # XXX, if shield logic *is wrong* inside
+        # XXX, if shield logic *is wrong* inside `open_root_actor()`'s
-        # `open_root_actor()`'s crash-handler block this should never
+        # crash-handler block this should never be interacted,
-        # be interacted, instead `trio.Cancelled` would be bubbled
+        # instead `trio.Cancelled` would be bubbled up: the original
-        # up: the original BUG.
+        # BUG.
         assert 0
 
 

examples/debugging/root_timeout_while_child_crashed.py

@@ -2,15 +2,12 @@ import trio
 import tractor
 
 
-async def key_error() -> None:
+async def key_error():
-    '''
+    "Raise a ``NameError``"
-    Raise a ``KeyError``.
-
-    '''
     return {}['doggy']
 
 
-async def main() -> None:
+async def main():
     '''
     Root is fail-after-cancelled while blocking and child RPC fails
     simultaneously.
@@ -24,7 +21,7 @@ async def main() -> None:
         trio.open_nursery() as tn,
     ):
         # spawn the actor..
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'key_error',
             enable_modules=[__name__],
         )
@@ -35,9 +32,9 @@ async def main() -> None:
         # root blocks below.
         tn.start_soon(portal.run, key_error)
 
-        # XXX: originally a bug caused by this is where root would
+        # XXX: originally a bug caused by this is where root would enter
-        # enter the debugger and clobber the tty used by the repl
+        # the debugger and clobber the tty used by the repl even though
-        # even though child should have it locked.
+        # child should have it locked.
         with trio.fail_after(1):
             await trio.Event().wait()
 

examples/debugging/shielded_pause.py

@@ -3,15 +3,8 @@ import tractor
 
 
 async def cancellable_pause_loop(
-    task_status: trio.TaskStatus[
+    task_status: trio.TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED
-        trio.CancelScope
+):
-    ] = trio.TASK_STATUS_IGNORED,
-) -> None:
-    '''
-    Exercise shielded debugger pauses under cancellation.
-
-    '''
-    cs: trio.CancelScope
     with trio.CancelScope() as cs:
         task_status.started(cs)
         for _ in range(3):
@@ -37,11 +30,7 @@ async def cancellable_pause_loop(
             await trio.lowlevel.checkpoint()
 
 
-async def pm_on_cancelled() -> None:
+async def pm_on_cancelled():
-    '''
-    Compare shielded and unshielded post-mortem entry.
-
-    '''
     async with trio.open_nursery() as tn:
         tn.cancel_scope.cancel()
         try:
@@ -67,7 +56,7 @@ async def pm_on_cancelled() -> None:
 
 
 async def cancelled_before_pause(
-) -> None:
+):
     '''
     Verify that using a shielded pause works despite surrounding
     cancellation called state in the calling task.
@@ -82,11 +71,7 @@ async def cancelled_before_pause(
     await pm_on_cancelled()
 
 
-async def main() -> None:
+async def main():
-    '''
-    Exercise shielded debugger entry in subactor and root tasks.
-
-    '''
     async with tractor.open_nursery(
         debug_mode=True,
     ) as an:

examples/debugging/subactor_bp_in_ctx.py

@@ -1,15 +1,10 @@
 import platform
-from collections.abc import AsyncIterator
 
 import tractor
 import trio
 
 
-async def gen() -> AsyncIterator[str]:
+async def gen():
-    '''
-    Yield values around debugger pauses.
-
-    '''
     yield 'yo'
     await tractor.pause()
     yield 'yo'
@@ -20,15 +15,11 @@ async def gen() -> AsyncIterator[str]:
 async def just_bp(
     ctx: tractor.Context,
 ) -> None:
-    '''
-    Pause repeatedly before deliberately breaking the context.
 
-    '''
     await ctx.started()
     await tractor.pause()
 
     # TODO: bps and errors in this call..
-    val: str
     async for val in gen():
         print(val)
 
@@ -43,18 +34,15 @@ async def just_bp(
 
 
 
-async def main() -> None:
+async def main():
-    '''
-    Run the breakpoint context over a supported transport.
 
-    '''
     # !TODO, parametrize the --tpt-proto={key} with osenv vars just
     # like we do for loglevel/spawn-backend!
     # - [ ] run on both tpts for all such debugger tests?
     # - [ ] special skip for macos!
     #
     if platform.system() != 'Darwin':
-        tpt: str = 'uds'
+        tpt = 'uds'
     else:
         # XXX, precisely we can't use pytest's tmp-path generation
         # for tests.. apparently because:
@@ -71,7 +59,7 @@ async def main() -> None:
         enable_transports=[tpt],
         loglevel='devx',
     ) as an:
-        p: tractor.Portal = await an.start_actor(
+        p = await an.start_actor(
             'bp_boi',
             enable_modules=[__name__],
         )

examples/debugging/subactor_breakpoint.py

@@ -2,7 +2,7 @@ import trio
 import tractor
 
 
-async def breakpoint_forever() -> None:
+async def breakpoint_forever():
     '''
     Indefinitely re-enter debugger in child actor.
 
@@ -12,11 +12,8 @@ async def breakpoint_forever() -> None:
         await tractor.pause()
 
 
-async def main() -> None:
+async def main():
-    '''
-    Run a subactor that repeatedly pauses in the debugger.
 
-    '''
     async with tractor.open_nursery(
         debug_mode=True,
         loglevel='cancel',

examples/debugging/subactor_error.py

@@ -2,19 +2,11 @@ import trio
 import tractor
 
 
-async def name_error() -> None:
+async def name_error():
-    '''
-    Deliberately raise a ``NameError`` in a subactor.
-
-    '''
     getattr(doggypants)  # noqa (on purpose)
 
 
-async def main() -> None:
+async def main():
-    '''
-    Surface a subactor `NameError` at the waiting root task.
-
-    '''
     async with tractor.open_nursery(
         debug_mode=True,
     ) as an:

examples/discovery/dedicated_registrar.py

@@ -1,394 +0,0 @@
-'''
-Run a dedicated registrar in a standalone process.
-
-The service and discovery client are sibling actors. The client has
-no pre-existing channel to the service, so its lookup must use the
-external registrar instead of the local-peer fast path.
-
-'''
-from __future__ import annotations
-
-from collections.abc import AsyncIterator
-from contextlib import asynccontextmanager as acm
-import errno
-from pathlib import Path
-import signal
-import socket
-import subprocess
-import sys
-import tempfile
-import time
-
-import trio
-import tractor
-
-
-MAX_BIND_ATTEMPTS: int = 5
-
-
-def _is_addr_collision(exc: BaseException) -> bool:
-    '''
-    Return whether registrar startup lost the selected TCP address.
-
-    Tractor can notice the collision while probing the address or
-    later when its listener binds. Exception groups are retryable
-    only when every contained failure reports the same collision.
-
-    '''
-    match exc:
-        case BaseExceptionGroup(exceptions=exceptions):
-            return bool(exceptions) and all(
-                _is_addr_collision(child)
-                for child in exceptions
-            )
-
-        case OSError() as os_error:
-            return (
-                os_error.errno in {errno.EADDRINUSE, 10048}
-                or getattr(os_error, 'winerror', None) == 10048
-            )
-
-        case RuntimeError() as runtime_error:
-            message: str = str(runtime_error)
-            return (
-                'Registry address(es) are occupied' in message
-                or 'registry socket(s) already bound' in message
-            )
-
-        case _:
-            return False
-
-
-def run_registrar(ready_path: str) -> None:
-    '''
-    Serve as the required registrar and report its selected address.
-
-    The kernel selects ephemeral loopback candidates in this process.
-    If another process claims a released candidate first, retry with
-    a fresh candidate up to `MAX_BIND_ATTEMPTS`. Other startup errors
-    and the final collision remain visible. `ensure_registry=True`
-    prevents silently joining a registrar that won the address.
-
-    '''
-    ready_file: Path = Path(ready_path)
-
-    async def serve() -> None:
-        '''
-        Open the registrar, publish readiness, and serve forever.
-
-        '''
-        for attempt in range(1, MAX_BIND_ATTEMPTS + 1):
-            # This selector socket reserves and reports a
-            # kernel-selected candidate; it never listens and is
-            # not transferred to Tractor. Closing it lets
-            # `open_root_actor()` create its own listener on the
-            # same addr. The close/rebind handoff is non-atomic,
-            # hence the bounded collision retries.
-            sock: socket.socket
-            with socket.socket(
-                socket.AF_INET,
-                socket.SOCK_STREAM,
-            ) as sock:
-                sock.bind(('127.0.0.1', 0))
-                selected: tuple[str, int] = sock.getsockname()
-                registry_addr: tuple[str, int] = (
-                    selected[0],
-                    selected[1],
-                )
-
-            try:
-                actor: tractor.Actor
-                async with tractor.open_root_actor(
-                    name='dedicated_registrar',
-                    registry_addrs=[registry_addr],
-                    enable_transports=['tcp'],
-                    enable_modules=[],
-                    ensure_registry=True,
-                    loglevel='error',
-                ) as actor:
-                    if not actor.is_registrar:
-                        raise RuntimeError(
-                            'daemon did not become registrar'
-                        )
-
-                    tmp_file: Path = ready_file.with_suffix('.tmp')
-                    tmp_file.write_text(
-                        str(registry_addr[1]),
-                        encoding='ascii',
-                    )
-                    tmp_file.replace(ready_file)
-                    await trio.sleep_forever()
-
-            except BaseException as exc:
-                if (
-                    not _is_addr_collision(exc)
-                    or attempt == MAX_BIND_ATTEMPTS
-                ):
-                    raise
-                await trio.sleep(.05 * attempt)
-
-    try:
-        trio.run(serve)
-    except KeyboardInterrupt:
-        pass
-
-
-def _registrar_command(ready_path: Path) -> list[str]:
-    '''
-    Build a child command that loads without running `main()`.
-
-    `runpy.run_path()` also works when the docs test copies and
-    renames this example before executing it.
-
-    '''
-    module_path: str = repr(str(Path(__file__).resolve()))
-    function_name: str = repr('run_registrar')
-    ready_arg: str = repr(str(ready_path))
-    code: str = (
-        f'import runpy; module = runpy.run_path({module_path}); '
-        f'module[{function_name}]({ready_arg})'
-    )
-    return [sys.executable, '-c', code]
-
-
-def _wait_registrar_ready(
-    ready_path: Path,
-    proc: subprocess.Popen,
-    deadline: float = 10.0,
-) -> tuple[str, int]:
-    '''
-    Wait until the child has entered its registrar actor context.
-
-    The child atomically publishes its selected port only after
-    `open_root_actor()` completes. Fail early if startup crashes.
-
-    '''
-    end: float = time.monotonic() + deadline
-    while time.monotonic() < end:
-        if proc.poll() is not None:
-            returncode: int|None = proc.returncode
-            raise RuntimeError(
-                f'registrar exited during startup: {returncode=}'
-            )
-
-        try:
-            port: int = int(
-                ready_path.read_text(encoding='ascii')
-            )
-        except (
-            OSError,
-            ValueError,
-        ):
-            time.sleep(.05)
-            continue
-
-        if not 0 < port < 2**16:
-            raise RuntimeError(f'invalid registrar port: {port!r}')
-        if proc.poll() is not None:
-            raise RuntimeError(
-                'registrar exited after reporting ready'
-            )
-        return ('127.0.0.1', port)
-
-    raise TimeoutError('registrar did not report ready')
-
-
-def _stop_registrar(
-    proc: subprocess.Popen,
-    graceful_timeout: float = 5.0,
-) -> None:
-    '''
-    Stop and reap the registrar, escalating after a bounded wait.
-
-    Windows children receive `CTRL_C_EVENT` in their new process
-    group; POSIX children receive `SIGINT`. A child that ignores
-    graceful shutdown is killed, and every path finishes with
-    `wait()`. A non-zero child exit remains visible to the caller.
-
-    '''
-    if proc.poll() is None:
-        graceful_signal: int = (
-            signal.CTRL_C_EVENT
-            if sys.platform == 'win32'
-            else signal.SIGINT
-        )
-        try:
-            proc.send_signal(graceful_signal)
-        except OSError:
-            if proc.poll() is None:
-                proc.terminate()
-
-    try:
-        proc.wait(timeout=graceful_timeout)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        proc.wait()
-
-    if proc.returncode:
-        raise RuntimeError(
-            'registrar shutdown failed: '
-            f'returncode={proc.returncode}'
-        )
-
-
-async def greet() -> str:
-    '''
-    Return a greeting identifying the actor serving the RPC.
-
-    '''
-    actor_name: str = tractor.current_actor().name
-    return f'hello from {actor_name}!'
-
-
-async def discover_and_greet(
-    registry_addr: tuple[str, int],
-) -> tuple[str, str, str]:
-    '''
-    Prove registrar lookup from a client without a service channel.
-
-    The parent spawns this actor as `greeter`'s sibling. A non-`None`
-    registry portal from `query_actor()` proves that discovery did
-    not take the existing-peer fast path, which returns no registry
-    portal.
-
-    '''
-    service_addr: tuple[str, int]|None
-    registry_portal: tractor.Portal|None
-    async with tractor.query_actor(
-        'greeter',
-        regaddr=registry_addr,
-    ) as (service_addr, registry_portal):
-        if registry_portal is None:
-            raise RuntimeError('lookup used a local service channel')
-        if service_addr is None:
-            raise RuntimeError('greeter was not registered')
-
-    service_portal: tractor.Portal|None
-    async with tractor.find_actor(
-        'greeter',
-        registry_addrs=[registry_addr],
-    ) as service_portal:
-        if service_portal is None:
-            raise RuntimeError('greeter disappeared before RPC')
-        greeting: str = await service_portal.run(greet)
-
-    client_name: str = tractor.current_actor().name
-    return client_name, repr(service_addr), greeting
-
-
-async def app(registry_addr: tuple[str, int]) -> None:
-    '''
-    Use sibling service and client actors with an external registrar.
-
-    Only the parent receives both spawn-time portals. The `client`
-    actor performs discovery in its own process and has no direct
-    `greeter` channel before the lookup.
-
-    '''
-    actor_nursery: tractor.ActorNursery
-    async with tractor.open_nursery(
-        registry_addrs=[registry_addr],
-        enable_transports=['tcp'],
-    ) as actor_nursery:
-        await actor_nursery.start_actor(
-            'greeter',
-            enable_modules=[__name__],
-        )
-        client_portal: tractor.Portal = (
-            await actor_nursery.start_actor(
-                'client',
-                enable_modules=[__name__],
-            )
-        )
-        result: tuple[str, str, str] = await client_portal.run(
-            discover_and_greet,
-            registry_addr=registry_addr,
-        )
-        client_name: str
-        service_addr: str
-        greeting: str
-        (
-            client_name,
-            service_addr,
-            greeting,
-        ) = result
-        print(
-            f'{client_name!r} found `greeter` through registrar '
-            f'{registry_addr!r}; service address: {service_addr}\n'
-            f'{greeting}'
-        )
-        await actor_nursery.cancel()
-
-
-# TODO: Promote this lifecycle into an OTB `tractor.discovery`
-# registrar subsystem. Reuse attach-or-create ownership from
-# `piker.service.maybe_open_pikerd()` and named service supervision
-# from `piker.service.Services`; replace the file readiness
-# handshake, then use the API from `tractor._testing.pytest` to
-# isolate remaining hard-coded `reg_addr` cases.
-@acm
-async def _open_registrar(
-) -> AsyncIterator[tuple[str, int]]:
-    '''
-    Start, publish, and reap one dedicated registrar process.
-
-    The Windows child gets a distinct console process group so the
-    graceful control event targets it without interrupting this
-    process.
-
-    '''
-    temp_dir: str
-    with tempfile.TemporaryDirectory(
-        prefix='tractor-registrar-',
-    ) as temp_dir:
-        ready_path: Path = Path(temp_dir) / 'ready'
-        creationflags: int = (
-            subprocess.CREATE_NEW_PROCESS_GROUP
-            if sys.platform == 'win32'
-            else 0
-        )
-        registrar: subprocess.Popen = subprocess.Popen(
-            _registrar_command(ready_path),
-            stdout=subprocess.DEVNULL,
-            creationflags=creationflags,
-        )
-        primary_error: BaseException|None = None
-        try:
-            registry_addr: tuple[str, int] = _wait_registrar_ready(
-                ready_path,
-                registrar,
-            )
-            print(
-                f'dedicated registrar ready at {registry_addr!r} '
-                f'(pid {registrar.pid})'
-            )
-            yield registry_addr
-        except BaseException as error:
-            primary_error = error
-            raise
-        finally:
-            try:
-                _stop_registrar(registrar)
-            except BaseException as cleanup_error:
-                if primary_error is None:
-                    raise
-                cleanup_note: str = (
-                    'registrar cleanup also failed: '
-                    f'{cleanup_error!r}'
-                )
-                primary_error.add_note(cleanup_note)
-            print('dedicated registrar shut down')
-
-
-async def main() -> None:
-    '''
-    Run the external registrar and sibling discovery actors.
-
-    '''
-    registry_addr: tuple[str, int]
-    async with _open_registrar() as registry_addr:
-        await app(registry_addr)
-
-
-if __name__ == '__main__':
-    trio.run(main)

examples/full_fledged_streaming_service.py

@@ -1,6 +1,4 @@
 import time
-from typing import AsyncIterator
-
 import trio
 import tractor
 from tractor import (
@@ -11,19 +9,14 @@ from tractor import (
 
 
 # this is the first 2 actors, streamer_1 and streamer_2
-async def stream_data(seed: int) -> AsyncIterator[int]:
+async def stream_data(seed):
-    '''
-    Stream integers up to a seed value.
-
-    '''
-    i: int
     for i in range(seed):
         yield i
         await trio.sleep(0.0001)  # trigger scheduler
 
 
 # this is the third actor; the aggregator
-async def aggregate(seed: int) -> AsyncIterator[int]:
+async def aggregate(seed):
     '''
     Ensure that the two streams we receive match but only stream
     a single set of values to the parent.
@@ -32,47 +25,30 @@ async def aggregate(seed: int) -> AsyncIterator[int]:
     an: ActorNursery
     async with tractor.open_nursery() as an:
         portals: list[Portal] = []
-        i: int
         for i in range(1, 3):
 
             # fork/spawn call
-            portal: Portal = await an.start_actor(
+            portal = await an.start_actor(
                 name=f'streamer_{i}',
                 enable_modules=[__name__],
             )
 
             portals.append(portal)
 
-        send_chan: trio.MemorySendChannel[int]
-        recv_chan: trio.MemoryReceiveChannel[int]
         send_chan, recv_chan = trio.open_memory_channel(500)
 
-        async def push_to_chan(
+        async def push_to_chan(portal, send_chan):
-            portal: Portal,
-            send_chan: trio.MemorySendChannel[int],
-        ) -> None:
-            '''
-            Forward one remote stream into a local channel.
 
-            '''
             # TODO: https://github.com/goodboy/tractor/issues/207
             async with send_chan:
-                stream: MsgStream
+                async with portal.open_stream_from(stream_data, seed=seed) as stream:
-                async with portal.open_stream_from(
-                    stream_data,
-                    seed=seed,
-                ) as stream:
-                    value: int
                     async for value in stream:
                         # leverage trio's built-in backpressure
                         await send_chan.send(value)
 
-            uid: tuple[str, str] = portal.chan.uid
+            print(f"FINISHED ITERATING {portal.channel.uid}")
-            print(f'FINISHED ITERATING {uid}')
 
-        # spawn 2 trio tasks to collect streams and push to a local
+        # spawn 2 trio tasks to collect streams and push to a local queue
-        # queue
-        n: trio.Nursery
         async with trio.open_nursery() as n:
 
             for portal in portals:
@@ -85,9 +61,8 @@ async def aggregate(seed: int) -> AsyncIterator[int]:
             # close this local task's reference to send side
             await send_chan.aclose()
 
-            unique_vals: set[int] = set()
+            unique_vals = set()
             async with recv_chan:
-                value: int
                 async for value in recv_chan:
                     if value not in unique_vals:
                         unique_vals.add(value)
@@ -96,11 +71,11 @@ async def aggregate(seed: int) -> AsyncIterator[int]:
 
                 assert value in unique_vals
 
-            print('FINISHED ITERATING in aggregator')
+            print("FINISHED ITERATING in aggregator")
 
         await an.cancel()
-        print('WAITING on `ActorNursery` to finish')
+        print("WAITING on `ActorNursery` to finish")
-    print('AGGREGATOR COMPLETE!')
+    print("AGGREGATOR COMPLETE!")
 
 
 async def main() -> list[int]:
@@ -119,8 +94,8 @@ async def main() -> list[int]:
         # debug_mode=True,
     ) as an:
 
-        seed: int = int(1e3)
+        seed = int(1e3)
-        pre_start: float = time.time()
+        pre_start = time.time()
 
         portal: Portal = await an.start_actor(
             name='aggregator',
@@ -133,27 +108,23 @@ async def main() -> list[int]:
             seed=seed,
         ) as stream:
 
-            start: float = time.time()
+            start = time.time()
             # the portal call returns exactly what you'd expect
-            # as if the remote "aggregate" function was called
+            # as if the remote "aggregate" function was called locally
-            # locally
             result_stream: list[int] = []
-            value: int
             async for value in stream:
                 result_stream.append(value)
 
         cancelled: bool = await portal.cancel_actor()
         assert cancelled
 
-        stream_time: float = time.time() - start
-        total_time: float = time.time() - pre_start
         print(
-            f'STREAM TIME = {stream_time}\n'
+            f"STREAM TIME = {time.time() - start}\n"
-            f'STREAM + SPAWN TIME = {total_time}\n'
+            f"STREAM + SPAWN TIME = {time.time() - pre_start}\n"
         )
         assert result_stream == list(range(seed))
         return result_stream
 
 
 if __name__ == '__main__':
-    final_stream: list[int] = trio.run(main)
+    final_stream = trio.run(main)

examples/infected_asyncio_echo_server.py

@@ -13,10 +13,7 @@ import tractor
 async def aio_echo_server(
     chan: tractor.to_asyncio.LinkedTaskChannel,
 ) -> None:
-    '''
-    Echo messages received through an asyncio task channel.
 
-    '''
     # a first message must be sent **from** this ``asyncio``
     # task or the ``trio`` side will never unblock from
     # ``tractor.to_asyncio.open_channel_from():``
@@ -31,15 +28,9 @@ async def aio_echo_server(
 @tractor.context
 async def trio_to_aio_echo_server(
     ctx: tractor.Context,
-) -> None:
+):
-    '''
-    Bridge an actor stream to the asyncio echo server.
-
-    '''
     # this will block until the ``asyncio`` task sends a "first"
     # message.
-    chan: tractor.to_asyncio.LinkedTaskChannel
-    first: str
     async with tractor.to_asyncio.open_channel_from(
         aio_echo_server,
     ) as (chan, first):
@@ -47,49 +38,39 @@ async def trio_to_aio_echo_server(
         assert first == 'start'
         await ctx.started(first)
 
-        stream: tractor.MsgStream
         async with ctx.open_stream() as stream:
 
-            msg: int
             async for msg in stream:
                 await chan.send(msg)
 
-                out: int = await chan.receive()
+                out = await chan.receive()
                 # echo back to parent actor-task
                 await stream.send(out)
 
 
-async def main() -> None:
+async def main():
-    '''
-    Run the infected asyncio echo-server example.
 
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        portal: tractor.Portal = await an.start_actor(
+        p = await an.start_actor(
             'aio_server',
             enable_modules=[__name__],
             infect_asyncio=True,
         )
-        ctx: tractor.Context
+        async with p.open_context(
-        first: str
-        async with portal.open_context(
             trio_to_aio_echo_server,
         ) as (ctx, first):
 
             assert first == 'start'
 
-            count: int = 0
+            count = 0
-            stream: tractor.MsgStream
             async with ctx.open_stream() as stream:
 
-                delays: list[float] = []
+                delays = []
-                send: float = time.time()
+                send = time.time()
 
                 await stream.send(count)
-                msg: int
                 async for msg in stream:
-                    recv: float = time.time()
+                    recv = time.time()
                     delays.append(recv - send)
                     assert msg == count
                     count += 1
@@ -100,7 +81,7 @@ async def main() -> None:
                         break
 
         print(f'mean round trip rate (Hz): {1/mean(delays)}')
-        await portal.cancel_actor()
+        await p.cancel_actor()
 
 
 if __name__ == '__main__':

examples/integration/mpi4py/inherit_parent_main.py

@@ -1,10 +1,9 @@
-'''
+"""
 Integration test: spawning tractor actors from an MPI process.
 
-When a parent is launched via ``mpirun``, Open MPI sets
+When a parent is launched via ``mpirun``, Open MPI sets ``OMPI_*`` env
-``OMPI_*`` env vars that bind ``MPI_Init`` to the ``orted``
+vars that bind ``MPI_Init`` to the ``orted`` daemon.  Tractor children
-daemon. Tractor children inherit those env vars, so if
+inherit those env vars, so if ``inherit_parent_main=True`` (the default)
-``inherit_parent_main=True`` (the default)
 the child re-executes ``__main__``, re-imports ``mpi4py``, and
 ``MPI_Init_thread`` fails because the child was never spawned by
 ``orted``::
@@ -13,15 +12,13 @@ the child re-executes ``__main__``, re-imports ``mpi4py``, and
       --> Returned value No permission (-17) instead of ORTE_SUCCESS
 
 Passing ``inherit_parent_main=False`` and placing RPC functions in a
-separate importable module (``_child``) avoids the re-import
+separate importable module (``_child``) avoids the re-import entirely.
-entirely.
 
 Usage::
 
     mpirun --allow-run-as-root -np 1 python -m \
         examples.integration.mpi4py.inherit_parent_main
-
+"""
-'''
 
 from mpi4py import MPI
 
@@ -33,27 +30,21 @@ from ._child import child_fn
 
 
 async def main() -> None:
-    '''
+    rank = MPI.COMM_WORLD.Get_rank()
-    Spawn an MPI-safe child without replaying the parent main.
+    print(f"[parent] rank={rank}  pid={os.getpid()}", flush=True)
 
-    '''
-    rank: int = MPI.COMM_WORLD.Get_rank()
-    pid: int = os.getpid()
-    print(f'[parent] rank={rank}  pid={pid}', flush=True)
-
-    an: tractor.ActorNursery
     async with tractor.open_nursery(start_method='trio') as an:
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'mpi-child',
             enable_modules=[child_fn.__module__],
             # Without this the child replays __main__, which
             # re-imports mpi4py and crashes on MPI_Init.
             inherit_parent_main=False,
         )
-        result: str = await portal.run(child_fn)
+        result = await portal.run(child_fn)
-        print(f'[parent] got: {result}', flush=True)
+        print(f"[parent] got: {result}", flush=True)
         await portal.cancel_actor()
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     trio.run(main)

examples/multiple_streams_one_portal.py

@@ -1,5 +1,3 @@
-from typing import AsyncIterator
-
 import trio
 import tractor
 
@@ -7,30 +5,17 @@ import tractor
 log = tractor.log.get_logger('multiportal')
 
 
-async def stream_data(seed: int = 10) -> AsyncIterator[int]:
+async def stream_data(seed=10):
-    '''
+    log.info("Starting stream task")
-    Stream a finite sequence of integers.
 
-    '''
-    log.info('Starting stream task')
-
-    i: int
     for i in range(seed):
         yield i
         await trio.sleep(0)  # trigger scheduler
 
 
-async def stream_from_portal(
+async def stream_from_portal(p, consumed):
-    portal: tractor.Portal,
-    consumed: list[int],
-) -> None:
-    '''
-    Consume one stream and toggle each value in a shared list.
 
-    '''
+    async with p.open_stream_from(stream_data) as stream:
-    stream: tractor.MsgStream
-    async with portal.open_stream_from(stream_data) as stream:
-        item: int
         async for item in stream:
             if item in consumed:
                 consumed.remove(item)
@@ -38,33 +23,20 @@ async def stream_from_portal(
                 consumed.append(item)
 
 
-async def main() -> None:
+async def main():
-    '''
-    Consume two concurrent streams through one portal.
 
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery(loglevel='info') as an:
 
-        portal: tractor.Portal = await an.start_actor(
+        p = await an.start_actor('stream_boi', enable_modules=[__name__])
-            'stream_boi',
-            enable_modules=[__name__],
-        )
 
-        consumed: list[int] = []
+        consumed = []
 
-        n: trio.Nursery
         async with trio.open_nursery() as n:
-            for _ in range(2):
+            for i in range(2):
-                n.start_soon(
+                n.start_soon(stream_from_portal, p, consumed)
-                    stream_from_portal,
-                    portal,
-                    consumed,
-                )
 
-        # both streaming consumer tasks have completed and so we
+        # both streaming consumer tasks have completed and so we should
-        # should have nothing in our list thanks to single
+        # have nothing in our list thanks to single threadedness
-        # threadedness
         assert not consumed
 
         await an.cancel()

examples/nested_actor_tree.py

@@ -41,7 +41,6 @@ async def fan_out_squares(
     aggregated squares to our parent.
 
     '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
         portals: list[tractor.Portal] = []
         for i in (1, 2):
@@ -53,28 +52,21 @@ async def fan_out_squares(
             )
         # unblock the parent's `.open_context()` entry and
         # report which leaves came up.
-        leaf_names: list[str] = [
+        await ctx.started(
-            portal.chan.aid.name
+            [p.chan.aid.name for p in portals]
-            for portal in portals
+        )
-        ]
-        await ctx.started(leaf_names)
         squares: dict[int, int] = {}
 
         async def run_in_leaf(
             portal: tractor.Portal,
             x: int,
         ) -> None:
-            '''
-            Run one square calculation in a leaf actor.
-
-            '''
             squares[x] = await portal.run(
                 compute_square,
                 x=x,
             )
 
         # fan out one sub-RPC per input val, concurrently.
-        tn: trio.Nursery
         async with trio.open_nursery() as tn:
             for i, x in enumerate(vals):
                 tn.start_soon(
@@ -91,18 +83,11 @@ async def fan_out_squares(
 
 
 async def main() -> None:
-    '''
-    Run the nested actor-tree example.
-
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'supervisor',
             enable_modules=[__name__],
         )
-        ctx: tractor.Context
-        leaf_names: list[str]
         async with portal.open_context(
             fan_out_squares,
             vals=[1, 2, 3, 4],

examples/parallelism/concurrent_actors_primes.py

@@ -1,23 +1,18 @@
-'''
+"""
 Demonstration of the prime number detector example from the
 ``concurrent.futures`` docs:
 
-https://docs.python.org/3/library/concurrent.futures.html\
+https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor-example
-#processpoolexecutor-example
 
 This uses no extra threads, fancy semaphores or futures; all we need
 is ``tractor``'s channels.
 
-'''
+"""
 from contextlib import (
     asynccontextmanager as acm,
     aclosing,
 )
-from typing import (
+from typing import Callable
-    AsyncIterator,
-    Awaitable,
-    Callable,
-)
 import itertools
 import math
 import time
@@ -26,12 +21,7 @@ import tractor
 import trio
 
 
-type ActorMap = Callable[
+PRIMES = [
-    [Callable[[int], Awaitable[bool]], list[int]],
-    AsyncIterator[tuple[int, bool]],
-]
-
-PRIMES: list[int] = [
     112272535095293,
     112582705942171,
     112272535095293,
@@ -41,11 +31,7 @@ PRIMES: list[int] = [
 ]
 
 
-async def is_prime(n: int) -> bool:
+async def is_prime(n):
-    '''
-    Return whether ``n`` is prime.
-
-    '''
     if n < 2:
         return False
     if n == 2:
@@ -61,32 +47,23 @@ async def is_prime(n: int) -> bool:
 
 
 @acm
-async def worker_pool(
+async def worker_pool(workers=4):
-    workers: int = 4,
+    """Though it's a trivial special case for ``tractor``, the well
-) -> AsyncIterator[ActorMap]:
-    '''
-    Though it's a trivial special case for ``tractor``, the well
     known "worker pool" seems to be the defacto "but, I want this
     process pattern!" for most parallelism pilgrims.
 
     Yes, the workers stay alive (and ready for work) until you close
     the context.
-
+    """
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
 
-        portals: list[tractor.Portal] = []
+        portals = []
-        snd_chan: trio.MemorySendChannel[tuple[int, bool]]
-        recv_chan: trio.MemoryReceiveChannel[tuple[int, bool]]
         snd_chan, recv_chan = trio.open_memory_channel(len(PRIMES))
 
-        i: int
         for i in range(workers):
 
-            # this starts a new sub-actor (process + trio
+            # this starts a new sub-actor (process + trio runtime) and
-            # runtime) and stores it's "portal" for later use to
+            # stores it's "portal" for later use to "submit jobs" (ugh).
-            # "submit jobs" (ugh).
             portals.append(
                 await an.start_actor(
                     f'worker_{i}',
@@ -95,36 +72,17 @@ async def worker_pool(
             )
 
         async def _map(
-            worker_func: Callable[[int], Awaitable[bool]],
+            worker_func: Callable[[int], bool],
-            sequence: list[int],
+            sequence: list[int]
-        ) -> AsyncIterator[tuple[int, bool]]:
+        ) -> list[bool]:
-            '''
-            Dispatch values across workers and yield their results.
 
-            '''
+            # define an async (local) task to collect results from workers
-            # define an async (local) task to collect results from
+            async def send_result(func, value, portal):
-            # workers
+                await snd_chan.send((value, await portal.run(func, n=value)))
-            async def send_result(
-                func: Callable[[int], Awaitable[bool]],
-                value: int,
-                portal: tractor.Portal,
-            ) -> None:
-                '''
-                Run one remote worker call and send its result.
 
-                '''
-                result: bool = await portal.run(func, n=value)
-                await snd_chan.send((value, result))
-
-            tn: trio.Nursery
             async with trio.open_nursery() as tn:
 
-                value: int
+                for value, portal in zip(sequence, itertools.cycle(portals)):
-                portal: tractor.Portal
-                for value, portal in zip(
-                    sequence,
-                    itertools.cycle(portals),
-                ):
                     tn.start_soon(
                         send_result,
                         worker_func,
@@ -143,30 +101,21 @@ async def worker_pool(
         await an.cancel()
 
 
-async def main() -> None:
+async def main():
-    '''
-    Report primality results from a pool of actors.
 
-    '''
-    actor_map: ActorMap
     async with worker_pool() as actor_map:
 
-        start: float = time.time()
+        start = time.time()
 
-        results: AsyncIterator[tuple[int, bool]]
         async with aclosing(actor_map(is_prime, PRIMES)) as results:
-            number: int
-            prime: bool
             async for number, prime in results:
 
                 print(f'{number} is prime: {prime}')
 
-        elapsed: float = time.time() - start
+        print(f'processing took {time.time() - start} seconds')
-        print(f'processing took {elapsed} seconds')
 
 
 if __name__ == '__main__':
-    start: float = time.time()
+    start = time.time()
     trio.run(main)
-    elapsed: float = time.time() - start
+    print(f'script took {time.time() - start} seconds')
-    print(f'script took {elapsed} seconds')

examples/parallelism/single_func.py

@@ -1,36 +1,30 @@
-'''
+"""
 Run with a process monitor from a terminal using::
 
     $TERM -e watch -n 0.1  "pstree -a $$" \
         & python examples/parallelism/single_func.py \
         && kill $!
 
-'''
+"""
 import os
 
 import tractor
 import trio
 
 
-async def burn_cpu() -> int:
+async def burn_cpu():
-    '''
-    Burn CPU briefly and return the current process ID.
 
-    '''
+    pid = os.getpid()
-    pid: int = os.getpid()
 
     # burn a core @ ~ 50kHz
     for _ in range(50000):
-        await trio.sleep(1 / 50000 / 50)
+        await trio.sleep(1/50000/50)
 
     return pid
 
 
-async def main() -> None:
+async def main():
-    '''
-    Run ``burn_cpu()`` in the parent and a subactor.
 
-    '''
     async with trio.open_nursery() as tn:
 
         # burn rubber in the parent too
@@ -38,9 +32,9 @@ async def main() -> None:
 
         # run the same func as the lone task in a subactor,
         # block on and collect its PID as the caller-side result
-        pid: int = await tractor.to_actor.run(burn_cpu)
+        pid = await tractor.to_actor.run(burn_cpu)
 
-    print(f'Collected subproc {pid}')
+    print(f"Collected subproc {pid}")
 
 
 if __name__ == '__main__':

examples/quick_cluster.py

@@ -1,24 +1,20 @@
+
 import trio
 import tractor
 
 
 async def sleepy_jane() -> None:
-    '''
+    uid: tuple = tractor.current_actor().uid
-    Identify the current actor and sleep forever.
-
-    '''
-    uid: tuple[str, str] = tractor.current_actor().uid
     print(f'Yo i am actor {uid}')
     await trio.sleep_forever()
 
 
-async def main() -> None:
+async def main():
     '''
     Spawn a flat actor cluster, with one process per detected core.
 
     '''
     portal_map: dict[str, tractor.Portal]
-    tn: trio.Nursery
 
     # look at this hip new syntax!
     async with (
@@ -31,9 +27,7 @@ async def main() -> None:
         trio.open_nursery() as tn,
     ):
 
-        name: str
+        for (name, portal) in portal_map.items():
-        portal: tractor.Portal
-        for name, portal in portal_map.items():
             tn.start_soon(
                 portal.run,
                 sleepy_jane,

examples/remote_error_propagation.py

@@ -2,30 +2,18 @@ import trio
 import tractor
 
 
-async def assert_err() -> None:
+async def assert_err():
-    '''
-    Raise an assertion error in the current actor.
-
-    '''
     assert 0
 
 
-async def main() -> None:
+async def main():
-    '''
-    Propagate a failing one-shot task from a subactor.
-
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        real_actors: list[tractor.Portal] = []
+        real_actors = []
-        i: int
         for i in range(3):
-            real_actors.append(
+            real_actors.append(await an.start_actor(
-                await an.start_actor(
                 f'actor_{i}',
                 enable_modules=[__name__],
-                )
+            ))
-            )
 
         # run one one-shot task actor that will fail immediately;
         # its error raises right here in the caller's task..
@@ -40,4 +28,4 @@ if __name__ == '__main__':
         # also raises
         trio.run(main)
     except tractor.RemoteActorError:
-        print('Look Maa that actor failed hard, hehhh!')
+        print("Look Maa that actor failed hard, hehhh!")

examples/rpc_bidir_streaming.py

@@ -9,19 +9,16 @@ async def simple_rpc(
     data: int,
 
 ) -> None:
-    '''
+    '''Test a small ping-pong 2-way streaming server.
-    Test a small ping-pong 2-way streaming server.
 
     '''
     # signal to parent that we're up much like
     # ``trio.TaskStatus.started()``
     await ctx.started(data + 1)
 
-    stream: tractor.MsgStream
     async with ctx.open_stream() as stream:
 
-        count: int = 0
+        count = 0
-        msg: str
         async for msg in stream:
 
             assert msg == 'ping'
@@ -33,22 +30,15 @@ async def simple_rpc(
 
 
 async def main() -> None:
-    '''
-    Exercise bidirectional streaming with a remote actor.
 
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
 
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'rpc_server',
             enable_modules=[__name__],
         )
 
         # XXX: syntax requires py3.9
-        ctx: tractor.Context
-        sent: int
-        stream: tractor.MsgStream
         async with (
 
             portal.open_context(
@@ -62,11 +52,10 @@ async def main() -> None:
 
             assert sent == 11
 
-            count: int = 0
+            count = 0
             # receive msgs using async for style
             await stream.send('ping')
 
-            msg: str
             async for msg in stream:
                 assert msg == 'pong'
                 await stream.send('ping')

examples/service_daemon_discovery.py

@@ -35,13 +35,11 @@ async def client_task() -> None:
 
     '''
     # a lookup miss yields `None` (not an error).
-    maybe_portal: tractor.Portal|None
+    async with tractor.find_actor('no_such_svc') as portal:
-    async with tractor.find_actor('no_such_svc') as maybe_portal:
+        assert portal is None
-        assert maybe_portal is None
         print('client: "no_such_svc" is not registered')
     # block until the service shows up in the registry,
     # then call into it through the delivered portal.
-    portal: tractor.Portal
     async with tractor.wait_for_actor('quote_svc') as portal:
         quote: float = await portal.run(
             get_quote,
@@ -51,19 +49,13 @@ async def client_task() -> None:
 
 
 async def main() -> None:
-    '''
-    Run a discoverable quote service and its client.
-
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'quote_svc',
             enable_modules=[__name__],
         )
         # run the client in a separate task which discovers
         # the daemon purely by its registered name.
-        tn: trio.Nursery
         async with trio.open_nursery() as tn:
             tn.start_soon(client_task)
         # explicit graceful teardown of the daemon.

examples/service_discovery.py

@@ -1,29 +1,19 @@
 import trio
 import tractor
 
-tractor.log.get_console_log('INFO')
+tractor.log.get_console_log("INFO")
 
 
-async def main(service_name: str) -> None:
+async def main(service_name):
-    '''
-    Discover one actor and inspect its registrar connection.
 
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
         await an.start_actor(service_name)
 
-        async with tractor.get_registry() as reg_portal:
+        async with tractor.get_registry() as portal:
-            print(
+            print(f"Registrar is listening on {portal.channel}")
-                f'Registrar is listening on {reg_portal.channel}'
-            )
 
-        actor_portal: tractor.Portal
+        async with tractor.wait_for_actor(service_name) as sockaddr:
-        async with tractor.wait_for_actor(
+            print(f"my_service is found at {sockaddr}")
-            service_name,
-        ) as actor_portal:
-            service_addr = actor_portal.chan.raddr
-            print(f'my_service is found at {service_addr}')
 
         await an.cancel()
 

examples/streaming_broadcast_fanout.py

@@ -28,7 +28,6 @@ async def tick_stream(
         # wait for the go-signal ensuring every parent-side
         # subscriber is attached before any tick is sent.
         assert await stream.receive() == 'go'
-        i: int
         for i in range(count):
             await stream.send(i)
         # falling out gracefully closes our stream side;
@@ -38,17 +37,15 @@ async def tick_stream(
 async def consume(
     name: str,
     stream: tractor.MsgStream,
-    task_status: trio.TaskStatus[None] = trio.TASK_STATUS_IGNORED,
+    task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED,
 ) -> None:
     '''
     Consume a private broadcast-copy of the IPC stream.
 
     '''
-    bcaster: tractor.trionics.BroadcastReceiver
     async with stream.subscribe() as bcaster:
         task_status.started()
         ticks: list[int] = []
-        tick: int
         async for tick in bcaster:
             print(f'{name}: rx {tick}')
             ticks.append(tick)
@@ -57,19 +54,11 @@ async def consume(
 
 
 async def main() -> None:
-    '''
-    Fan one remote stream out to local subscribers.
-
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'ticker',
             enable_modules=[__name__],
         )
-        ctx: tractor.Context
-        first: int
-        stream: tractor.MsgStream
         async with (
             portal.open_context(
                 tick_stream,
@@ -78,11 +67,9 @@ async def main() -> None:
             ctx.open_stream() as stream,
         ):
             assert first == 5
-            tn: trio.Nursery
             async with trio.open_nursery() as tn:
                 # use `.start()` so each consumer is known
                 # to be subscribed before the ticks flow.
-                i: int
                 for i in range(3):
                     await tn.start(
                         consume,

examples/trio/lockacquire_not_unmasked.py

@@ -1,4 +1,3 @@
-from collections.abc import AsyncIterator
 from contextlib import (
     asynccontextmanager as acm,
 )
@@ -17,11 +16,7 @@ _lock: trio.Lock|None = None
 
 @acm
 async def acquire_singleton_lock(
-) -> AsyncIterator[trio.Lock]:
+) -> None:
-    '''
-    Acquire and yield the process-wide lock.
-
-    '''
     global _lock
     if _lock is None:
         log.info('Allocating LOCK')
@@ -37,15 +32,8 @@ async def acquire_singleton_lock(
 
 
 async def hold_lock_forever(
-    task_status: trio.TaskStatus[
+    task_status=trio.TASK_STATUS_IGNORED
-        trio.Lock,
+):
-    ] = trio.TASK_STATUS_IGNORED,
-) -> None:
-    '''
-    Hold the singleton lock until cancellation.
-
-    '''
-    lock: trio.Lock
     async with (
         tractor.trionics.maybe_raise_from_masking_exc(),
         acquire_singleton_lock() as lock,
@@ -58,12 +46,7 @@ async def main(
     ignore_special_cases: bool,
     loglevel: str = 'info',
     debug_mode: bool = True,
-) -> None:
+):
-    '''
-    Exercise lock acquisition while cancellation is masked.
-
-    '''
-    tn: trio.Nursery
     async with (
         trio.open_nursery() as tn,
 
@@ -75,7 +58,7 @@ async def main(
             from tractor.trionics import _taskc
             _taskc._mask_cases.clear()
 
-        _held_lock: trio.Lock = await tn.start(
+        _lock = await tn.start(
             hold_lock_forever,
         )
         with trio.move_on_after(0.2):
@@ -91,8 +74,8 @@ if __name__ == '__main__':
     tractor.log.get_console_log(level='info')
     for case in [True, False]:
         log.info(
-            '\n'
+            f'\n'
-            '------ RUNNING SCRIPT TRIAL ------\n'
+            f'------ RUNNING SCRIPT TRIAL ------\n'
             f'ignore_special_cases: {case!r}\n'
         )
         trio.run(partial(

examples/trio/send_chan_aclose_masks_beg.py

@@ -1,4 +1,3 @@
-from collections.abc import Iterator
 from contextlib import (
     contextmanager as cm,
     # TODO, any diff in async case(s)??
@@ -18,7 +17,7 @@ log = tractor.log.get_logger(
 @cm
 def teardown_on_exc(
     raise_from_handler: bool = False,
-) -> Iterator[None]:
+):
     '''
     You could also have a teardown handler which catches any exc and
     does some required teardown. In this case the problem is
@@ -31,7 +30,7 @@ def teardown_on_exc(
     except BaseException as _berr:
         berr = _berr
         log.exception(
-            'Handling termination teardown in child due to,\n'
+            f'Handling termination teardown in child due to,\n'
             f'{berr!r}\n'
         )
         if raise_from_handler:
@@ -55,18 +54,14 @@ def teardown_on_exc(
 
 
 async def finite_stream_to_rent(
-    tx: trio.abc.SendChannel[int],
+    tx: trio.abc.SendChannel,
     child_errors_mid_stream: bool,
     raise_unmasked: bool,
 
     task_status: trio.TaskStatus[
-        trio.CancelScope|None,
+        trio.CancelScope,
     ] = trio.TASK_STATUS_IGNORED,
-) -> None:
+):
-    '''
-    Stream values while reproducing exception masking on close.
-
-    '''
     async with (
         # XXX without this unmasker the mid-streaming RTE is never
         # reported since it is masked by the `tx.aclose()`
@@ -139,26 +134,19 @@ async def main(
 
     raise_unmasked: bool = False,
     loglevel: str = 'info',
-) -> None:
+):
-    '''
-    Reproduce cancellation masking a child-stream failure.
-
-    '''
     tractor.log.get_console_log(level=loglevel)
 
     # the `.aclose()` being checkpoints on these
     # is the source of the problem..
-    tx: trio.MemorySendChannel[int]
-    rx: trio.MemoryReceiveChannel[int]
     tx, rx = trio.open_memory_channel(1)
 
-    tn: trio.Nursery
     async with (
         tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn,
         rx as rx,
     ):
-        _child_cs: trio.CancelScope|None = await tn.start(
+        _child_cs = await tn.start(
             partial(
                 finite_stream_to_rent,
                 child_errors_mid_stream=child_errors_mid_stream,
@@ -166,7 +154,6 @@ async def main(
                 tx=tx,
             )
         )
-        msg: int
         async for msg in rx:
             log.debug(
                 f'Rent rx {msg!r}\n'
@@ -175,13 +162,12 @@ async def main(
             # simulate some external cancellation
             # request **JUST BEFORE** the child errors.
             if msg == 65:
-                cancel_msg: str = (
+                log.cancel(
-                    'Cancelling parent on,\n'
+                    f'Cancelling parent on,\n'
-                    'msg={msg}\n'
+                    f'msg={msg}\n'
-                    '\n'
+                    f'\n'
-                    'Simulates OOB cancel request!\n'
+                    f'Simulates OOB cancel request!\n'
-                ).format(msg=msg)
+                )
-                log.cancel(cancel_msg)
                 tn.cancel_scope.cancel()
 
 
@@ -190,8 +176,8 @@ if __name__ == '__main__':
     tractor.log.get_console_log(level='info')
     for case in [True, False]:
         log.info(
-            '\n'
+            f'\n'
-            '------ RUNNING SCRIPT TRIAL ------\n'
+            f'------ RUNNING SCRIPT TRIAL ------\n'
             f'child_errors_midstream: {case!r}\n'
         )
         try:

examples/typed_payloads.py

@@ -46,9 +46,7 @@ async def point_doubler(
     # now do it right; the parent receives this as the 2nd
     # element of its `.open_context()` entry tuple.
     await ctx.started(Point(x=0, y=0))
-    stream: tractor.MsgStream
     async with ctx.open_stream() as stream:
-        pt: Point
         async for pt in stream:
             # natively decoded to our struct type!
             assert type(pt) is Point
@@ -61,19 +59,11 @@ async def point_doubler(
 
 
 async def main() -> None:
-    '''
-    Exchange typed ``Point`` payloads with a subactor.
-
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'point_doubler',
             enable_modules=[__name__],
         )
-        ctx: tractor.Context
-        first: Point
-        stream: tractor.MsgStream
         async with (
             portal.open_context(
                 point_doubler,
@@ -82,7 +72,6 @@ async def main() -> None:
         ):
             # the (validated) started-value from the child
             assert first == Point(x=0, y=0)
-            i: int
             for i in range(3):
                 await stream.send(Point(x=i, y=i))
                 doubled: Point = await stream.receive()

examples/uds_transport_actor_tree.py

@@ -21,22 +21,17 @@ async def report_addr() -> str:
     Return this actor's own accept (bind) addr + pid.
 
     '''
-    actor: tractor.Actor = tractor.current_actor()
+    actor = tractor.current_actor()
-    addr: tuple[str, int|str] = actor.accept_addr
+    addr: tuple = actor.accept_addr
     pid: int = os.getpid()
     return f'{actor.name}@{addr} pid={pid}'
 
 
 async def main() -> None:
-    '''
-    Run a child actor over the UDS transport.
-
-    '''
-    an: tractor.ActorNursery
     async with tractor.open_nursery(
         enable_transports=['uds'],
     ) as an:
-        portal: tractor.Portal = await an.start_actor(
+        portal = await an.start_actor(
             'uds_child',
             enable_modules=[__name__],
         )
@@ -56,8 +51,7 @@ async def main() -> None:
         )
         # ask the child for its OWN distinct bind addr: another
         # socket-file path under the runtime dir.
-        child_report: str = await portal.run(report_addr)
+        print(f'child says: {await portal.run(report_addr)}')
-        print(f'child says: {child_report}')
         await portal.cancel_actor()