Pass `Bindspace` FD to `trio` children

Thread an optional live `Bindspace` through actor spawn APIs and
give each Trio exec child an inherited namespace descriptor before
runtime bootstrap.

Deats,
- duplicate the namespace FD without changing parent ownership
- preserve caller `pass_fds` and process options
- send the `(fd, inode)` pair through the `_child` CLI
- close the parent duplicate on success, failure and cancellation
- reject MP backends until descriptor reduction is implemented
- avoid cleanup before child publication
- exercise real E2E namespace relay through handshake and RPC
- verify parent FD-table cleanup after successful child spawn

Prompt-IO: ai/prompt-io/opencode/20260828T172943Z_2ca8c570_prompt_io.md

(this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`))

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

@@ -0,0 +1,62 @@
+---
+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

@@ -0,0 +1,90 @@
+---
+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. Name-based lookups simply
-   resolve to the *last* registered match, so if you boot five
-   actors all named ``'bob'``, you get the freshest ``'bob'`` B)
+   by the full ``(name, uuid)`` pair. A name lookup returns one
+   matching registration, but the API does not promise which match
+   wins. Use unique service names when selection matters.
 
 First boot: who's the registrar?
 --------------------------------
@@ -65,6 +65,43 @@ 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
 -----------------
 
@@ -92,13 +129,22 @@ Knobs worth knowing:
 - ``registry_addrs=[...]``: query specific (possibly multiple,
   possibly remote) registrars instead of your tree's default,
 
-- ``only_first=False``: deliver a ``list[Portal]`` of *all*
-  matches found across the queried registrars instead of just the
-  first,
+- ``only_first=True``: after all configured registrars are queried
+  concurrently, yield the result in the first ``registry_addrs``
+  position. This is configured order, not first-reachable order, so
+  the result can be ``None`` even when a later registrar returned a
+  portal,
 
-- ``raise_on_none=True``: raise a ``RuntimeError`` instead of
-  yielding ``None`` when no match is found — for when absence is
-  a hard error in your app.
+- ``only_first=False``: when any query succeeds, yield an ordered
+  ``list[Portal | None]`` with one result per ``registry_addrs``
+  position; misses remain ``None`` placeholders. When every query
+  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()``
 ********************
@@ -131,10 +177,11 @@ Yields a portal straight to the registrar actor itself — or a
 Fast paths and address preference
 ---------------------------------
 
-Before doing any RPC to the registrar, every lookup first scans
-the calling actor's *already-connected peers*: if you have a live
-channel to an actor named ``name`` you get a portal over it
-immediately, no registrar round-trip at all.
+Before doing any RPC to the registrar, ``query_actor()``,
+``wait_for_actor()``, and the default ``find_actor()`` lookup first
+scan the calling actor's *already-connected peers*. If the caller
+has a live channel to an actor named ``name``, it gets a portal over
+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:
@@ -201,8 +248,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 Tractor
-registrar handshake fails startup instead of being rebound.
+as a tree-joiner. An occupied address that does not complete a
+Tractor registrar handshake fails startup instead of being rebound.
 
 "Arbiter"? A legacy naming note
 -------------------------------
@@ -226,10 +273,11 @@ Very naive, very honest
 -----------------------
 
 To be clear, this is a **very naive** discovery system: one
-process-tree-local registrar holding a dict, no replication, no
-re-election when it dies, no cross-host propagation. That's
-intentional (for now); it covers the "wire up my services on this
-host" case without dragging in a consensus protocol.
+in-memory registrar holding a dict, no replication, no re-election
+when it dies, and no automatic cross-host propagation. Separate
+programs can use the same reachable registrar, as above, but must be
+configured with its address. That's intentional (for now); it covers
+the "wire up my services" case without a consensus protocol.
 
 On the roadmap (issue `#216`_ tracks a chunk of it):
 
@@ -254,6 +302,7 @@ to hear from you.
      :class:`tractor.Registrar`.
 
 .. _gossip protocol: https://en.wikipedia.org/wiki/Gossip_protocol
-.. _modern protocol: https://en.wikipedia.org/wiki/Rendezvous_protocol
+.. _modern 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,14 +153,16 @@ 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`` payloads are **never** checked inside
-  ``MsgStream.send()``; they're validated receiver-side on each
-  ``MsgStream.receive()``. A violation raises a ``MsgTypeError``
-  in the receiver *and* relays an ``Error`` msg back so the
-  offending sender gets one raised too.
-- the remaining control msgs (``Start``, ``Return``) are likewise
-  validated such that violations raise in the **sending** actor,
-  pointing the traceback at the code that actually goofed.
+- ``Yield`` and ``Return`` payloads are not checked before sending;
+  they're decoded against the dialog's spec by the receiver. A
+  violation raises a ``MsgTypeError`` there and terminates that
+  dialog. The peer then observes the resulting protocol teardown;
+  it is not guaranteed to receive the same ``MsgTypeError``.
+- ``Start`` arguments are dispatched through the RPC endpoint's
+  Python signature. They are not payloads covered by the dialog's
+  ``pld_spec``. A planned follow-up will derive a typed ``Start``
+  contract from endpoint annotations and validate arguments
+  sender-side; see `#514`_.
 
 Anatomy of a ``MsgTypeError``
 -----------------------------
@@ -177,21 +179,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 (``Started``,
-``Return``) points straight at your offending ``await
-ctx.started()`` or ``return`` statement, while a *receiver-side*
-MTE (``Yield``) surfaces from the consumer's ``receive()`` call
-with the relay copy delivered back to the producer. Either way
-the failure is scoped to that one dialog; sibling contexts on the
-same channel keep right on trucking.
+Practical reading guide: a *sender-side* MTE for ``Started`` points
+straight at the offending ``await ctx.started()`` call. A
+*receiver-side* MTE for ``Yield`` or ``Return`` surfaces while the
+peer decodes the payload. Either way 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
-easiest path is per-endpoint: ``@tractor.context()`` accepts
-``enc_hook``/``dec_hook`` params right alongside ``pld_spec``.
-For full control build and apply a codec yourself; encode-side:
+anything else you teach the codec via extension hooks. The complete
+public path currently available is task-scoped encoding:
+``tractor.msg.mk_codec()`` builds a codec with an ``enc_hook``, and
+``tractor.msg.apply_codec()`` installs it for the current task. To
+build and apply that transport codec:
 
 __ https://jcristharif.com/msgspec/supported-types.html
 
@@ -206,8 +208,9 @@ __ https://jcristharif.com/msgspec/supported-types.html
     with apply_codec(codec):        # ContextVar-scoped override
         ...  # msgs sent by this task now encode NSPs
 
-and decode-side, scoped to an open context (note the import from
-``tractor.msg._ops``, not yet re-exported):
+The context manager which temporarily installs payload-decoder
+settings on an open context is separate and still private (note
+the ``tractor.msg._ops`` import):
 
 .. code:: python
 
@@ -220,11 +223,15 @@ and decode-side, scoped to an open context (note the import from
     ):
         ...  # this dialog's payloads decode as NSPs
 
-``apply_codec()`` is ``ContextVar``-scoped: it overrides the
-codec for the current task (and only that task), not the whole
-process. For complete working flows, including hook pairing rules
-and roundtrip cases, see ``tests/msg/test_ext_types_msgspec.py``
-and ``tests/msg/test_pldrx_limiting.py``.
+``apply_codec()`` is ``ContextVar``-scoped: it overrides the codec
+for the current task (and only that task), not the whole process.
+``@tractor.context()`` accepts ``enc_hook`` and ``dec_hook``
+parameters, but their runtime wiring is not yet a symmetric,
+end-to-end public hook pair: the encode hook is not consumed and
+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
@@ -238,9 +245,33 @@ 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. That work is tracked in `#196`_ (with the
-original typed-proto epic in `#36`_); if strongly-typed
-distributed systems get you going, we'd love your input.
+of the protocol itself. The epic is tracked in `#196`_ (evolving
+the original typed-proto work in `#36`_), and the most recent
+concrete step is `#365`_ — driving the whole ``pld_spec`` off
+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?
 --------------
@@ -258,3 +289,6 @@ 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,28 +1,40 @@
 import trio
 import tractor
 
-_this_module = __name__
-the_line = 'Hi my name is {}'
+_this_module: str = __name__
+the_line: str = 'Hi my name is {}'
 
 
-tractor.log.get_console_log("INFO")
+tractor.log.get_console_log('INFO')
 
 
-async def hi():
+async def hi() -> str:
+    '''
+    Return a greeting naming the current actor.
+
+    '''
     return the_line.format(tractor.current_actor().name)
 
 
-async def say_hello(other_actor):
+async def say_hello(other_actor: str) -> str:
+    '''
+    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():
-    """Main tractor entry point, the "master" process (for now
+async def main() -> None:
+    '''
+    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
@@ -39,6 +51,10 @@ async def main():
             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(
@@ -47,13 +63,14 @@ async def main():
                 )
             )
 
+        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,14 +2,20 @@ import trio
 import tractor
 
 
-async def cellar_door():
+async def cellar_door() -> str:
+    '''
+    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():
-    """The main ``tractor`` routine.
-    """
+async def main() -> None:
+    '''
+    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,19 +2,24 @@ import trio
 import tractor
 
 
-async def movie_theatre_question():
-    """A question asked in a dark theatre, in a tangent
+async def movie_theatre_question() -> str:
+    '''
+    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():
-    """The main ``tractor`` routine.
-    """
+async def main() -> None:
+    '''
+    The main ``tractor`` routine.
+
+    '''
+    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
 
-        portal = await an.start_actor(
+        portal: tractor.Portal = await an.start_actor(
             'frank',
             # enable the actor to run funcs from this current module
             enable_modules=[__name__],
@@ -24,9 +29,8 @@ async def main():
         # call the subactor a 2nd time
         print(await portal.run(movie_theatre_question))
 
-        # the async with will block here indefinitely waiting
-        # for our actor "frank" to complete, but since it's an
-        # "outlive_main" actor it will never end until cancelled
+        # the async with will wait indefinitely for "frank" because
+        # its runtime remains active until explicitly cancelled
         await portal.cancel_actor()
 
 

examples/asynchronous_generators.py

@@ -1,33 +1,46 @@
-from typing import AsyncIterator
 from itertools import repeat
+from typing import AsyncIterator
 
 import trio
 import tractor
 
 
-async def stream_forever() -> AsyncIterator[int]:
+async def stream_forever() -> AsyncIterator[str]:
+    '''
+    Stream the same message indefinitely.
 
-    for i in repeat("I can see these little future bubble things"):
-        # each yielded value is sent over the ``Channel`` to the parent actor
-        yield i
+    '''
+    message: str
+    for message in repeat(
+        '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():
+async def main() -> None:
+    '''
+    Print messages streamed from a subactor.
 
+    '''
+    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
 
-        portal = await an.start_actor(
+        portal: tractor.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 = 0
-            async for letter in stream:
-                print(letter)
+            count: int = 0
+            message: str
+            async for message in stream:
+                print(message)
                 count += 1
 
                 if count > 50:

examples/debugging/fast_error_in_root_after_spawn.py

@@ -13,7 +13,11 @@ 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()
@@ -21,10 +25,13 @@ 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 = await n.start_actor(
+    portal: tractor.Portal = await n.start_actor(
         name='sleeper',
         enable_modules=[__name__],
     )
@@ -35,8 +42,11 @@ async def open_ctx(
         assert first is None
 
 
-async def main():
+async def main() -> None:
+    '''
+    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,9 +1,14 @@
+from collections.abc import AsyncIterator
+
 import tractor
 import trio
 
 
-async def breakpoint_forever():
-    "Indefinitely re-enter debugger in child actor."
+async def breakpoint_forever() -> AsyncIterator[str]:
+    '''
+    Indefinitely re-enter debugger in child actor.
+
+    '''
     try:
         while True:
             yield 'yo'
@@ -15,12 +20,15 @@ async def breakpoint_forever():
         raise
 
 
-async def name_error():
-    "Raise a ``NameError``"
+async def name_error() -> None:
+    '''
+    Raise a ``NameError``.
+
+    '''
     getattr(doggypants)  # noqa
 
 
-async def main():
+async def main() -> None:
     '''
     Test breakpoint in a streaming actor.
 
@@ -28,8 +36,14 @@ async def main():
     async with tractor.open_nursery(
         debug_mode=True,
     ) as an:
-        p0 = await an.start_actor('bp_forever', enable_modules=[__name__])
-        p1 = await an.start_actor('name_error', enable_modules=[__name__])
+        p0: tractor.Portal = await an.start_actor(
+            'bp_forever',
+            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:
@@ -40,6 +54,7 @@ async def main():
             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,13 +4,19 @@ import trio
 import tractor
 
 
-async def name_error():
-    "Raise a ``NameError``"
+async def name_error() -> None:
+    '''
+    Raise a ``NameError``.
+
+    '''
     getattr(doggypants)  # noqa
 
 
-async def breakpoint_forever():
-    "Indefinitely re-enter debugger in child actor."
+async def breakpoint_forever() -> None:
+    '''
+    Indefinitely re-enter debugger in child actor.
+
+    '''
     while True:
         await tractor.pause()
 
@@ -20,9 +26,13 @@ async def breakpoint_forever():
         # await trio.sleep(0)
 
 
-async def spawn_until(depth=0):
-    """"A nested nursery that triggers another ``NameError``.
-    """
+async def spawn_until(
+    depth: int = 0,
+) -> None:
+    '''
+    A nested nursery that triggers another ``NameError``.
+
+    '''
     async with (
         tractor.open_nursery() as an,
         trio.open_nursery() as tn,
@@ -37,7 +47,8 @@ async def spawn_until(depth=0):
                 )
             )
 
-            # Let the background one-shot enter `breakpoint_forever()`
+            # Let the background one-shot enter
+            # `breakpoint_forever()`
             # before its sibling raises and cancellation propagates.
             await trio.sleep(0.5)
             # rx and propagate error from child
@@ -48,9 +59,9 @@ async def spawn_until(depth=0):
             )
 
         else:
-            # recusrive call to spawn another process branching layer of
-            # the tree; blocks (up) each level until the leaf's
-            # `name_error` relays through.
+            # recusrive call to spawn another process branching
+            # layer of the tree; blocks (up) each level until the
+            # leaf's `name_error` relays through.
             depth -= 1
             await tractor.to_actor.run(
                 partial(
@@ -63,25 +74,35 @@ async def spawn_until(depth=0):
 
 
 # TODO: notes on the new boxed-relayed errors through proxy actors
-async def main():
-    """The main ``tractor`` routine.
+async def main() -> None:
+    '''
+    The main ``tractor`` routine.
 
-    The process tree should look as approximately as follows when the debugger
-    first engages:
+    The process tree should look approximately as follows when the
+    debugger first engages:
 
     python examples/debugging/multi_nested_subactors_bp_forever.py
-    ├─ python -m tractor._child --uid ('spawner1', '7eab8462 ...)
-    │  └─ python -m tractor._child --uid ('spawn_until_3', 'afcba7a8 ...)
-    │     └─ python -m tractor._child --uid ('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
+    │  ('spawner1', '7eab8462 ...')
+    │  └─ python -m tractor._child --uid
+    │     ('spawn_until_3', 'afcba7a8 ...')
+    │     └─ python -m tractor._child --uid
+    │        ('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 ('spawner0', '1d42012b ...)
-       └─ python -m tractor._child --uid ('spawn_until_2', '2877e155 ...)
-          └─ python -m tractor._child --uid ('spawn_until_1', '0502d786 ...)
-             └─ python -m tractor._child --uid ('spawn_until_0', 'de918e6d ...)
+    └─ python -m tractor._child --uid
+       ('spawner0', '1d42012b ...')
+       └─ python -m tractor._child --uid
+          ('spawn_until_2', '2877e155 ...')
+          └─ 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,14 +7,19 @@ import trio
 import tractor
 
 
-async def name_error():
-    "Raise a ``NameError``"
+async def name_error() -> None:
+    '''
+    Raise a ``NameError``.
+
+    '''
     getattr(doggypants)  # noqa
 
 
-async def spawn_error():
-    """"A nested nursery that triggers another ``NameError``.
-    """
+async def spawn_error() -> None:
+    '''
+    A nested nursery that triggers another ``NameError``.
+
+    '''
     async with tractor.open_nursery() as an:
         return await tractor.to_actor.run(
             name_error,
@@ -23,8 +28,9 @@ async def spawn_error():
         )
 
 
-async def main():
-    """The main ``tractor`` routine.
+async def main() -> None:
+    '''
+    The main ``tractor`` routine.
 
     The process tree should look as approximately as follows:
 
@@ -37,7 +43,8 @@ async def main():
         - nested name_error sub-sub-actor
         - root actor should then fail on assert
         - program termination
-    """
+
+    '''
     async with (
         tractor.open_nursery(
             debug_mode=True,
@@ -46,11 +53,11 @@ async def main():
         trio.open_nursery() as tn,
     ):
         # spawn both actors..
-        portal = await an.start_actor(
+        portal: tractor.Portal = await an.start_actor(
             'name_error',
             enable_modules=[__name__],
         )
-        portal1 = await an.start_actor(
+        portal1: tractor.Portal = await an.start_actor(
             'spawn_error',
             enable_modules=[__name__],
         )

examples/debugging/multi_subactors.py

@@ -1,22 +1,32 @@
+from collections.abc import Awaitable, Callable
+
 import tractor
 import trio
 
 
-async def breakpoint_forever():
-    "Indefinitely re-enter debugger in child actor."
+async def breakpoint_forever() -> None:
+    '''
+    Indefinitely re-enter debugger in child actor.
+
+    '''
     while True:
         await trio.sleep(0.1)
         await tractor.pause()
 
 
-async def name_error():
-    "Raise a ``NameError``"
+async def name_error() -> None:
+    '''
+    Raise a ``NameError``.
+
+    '''
     getattr(doggypants)  # noqa
 
 
-async def spawn_error():
-    """"A nested nursery that triggers another ``NameError``.
-    """
+async def spawn_error() -> None:
+    '''
+    A nested nursery that triggers another ``NameError``.
+
+    '''
     async with tractor.open_nursery() as an:
         return await tractor.to_actor.run(
             name_error,
@@ -25,8 +35,9 @@ async def spawn_error():
         )
 
 
-async def main():
-    """The main ``tractor`` routine.
+async def main() -> None:
+    '''
+    The main ``tractor`` routine.
 
     The process tree should look as approximately as follows:
 
@@ -35,7 +46,8 @@ async def main():
     |-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(
@@ -43,12 +55,14 @@ async def main():
         # loglevel='runtime',
     ) as an:
 
-        async def run_and_collect(fn):
+        async def run_and_collect(
+            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,19 +1,28 @@
 import trio
 import tractor
 
-async def die():
+
+async def die() -> None:
+    '''
+    Deliberately crash the calling actor.
+
+    '''
     raise RuntimeError
 
 
-async def main():
+async def main() -> None:
+    '''
+    Crash actors with different debugger settings concurrently.
+
+    '''
     async with tractor.open_nursery() as an:
 
-        debug_actor = await an.start_actor(
+        debug_actor: tractor.Portal = await an.start_actor(
             'debugged_boi',
             enable_modules=[__name__],
             debug_mode=True,
         )
-        crash_boi = await an.start_actor(
+        crash_boi: tractor.Portal = 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():
+async def main() -> None:
     '''
     Test 3 `PdbREPL` entries:
       - one in the child due to manual `.post_mortem()`,
@@ -49,7 +49,9 @@ async def main():
             await tractor.post_mortem()
             raise
         else:
-            raise RuntimeError('IPC ctx should have remote errored!?')
+            raise RuntimeError(
+                'IPC ctx should have remote errored!?'
+            )
 
 
 if __name__ == '__main__':

examples/debugging/root_actor_breakpoint.py

@@ -2,8 +2,11 @@ import trio
 import tractor
 
 
-async def main():
+async def main() -> None:
+    '''
+    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,7 +2,11 @@ import trio
 import tractor
 
 
-async def main():
+async def main() -> None:
+    '''
+    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,14 +4,21 @@ import trio
 import tractor
 
 
-async def name_error():
-    "Raise a ``NameError``"
+async def name_error() -> None:
+    '''
+    Raise a ``NameError``.
+
+    '''
     getattr(doggypants)  # noqa
 
 
-async def spawn_until(depth=0):
-    """"A nested nursery that triggers another ``NameError``.
-    """
+async def spawn_until(
+    depth: int = 0,
+) -> 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)
@@ -27,18 +34,23 @@ async def spawn_until(depth=0):
             )
 
 
-async def main():
+async def main() -> None:
     '''
     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 ('spawner1', '7eab8462 ...)
-    │  └─ python -m tractor._child --uid ('spawn_until_0', '3720602b ...)
-    │     └─ python -m tractor._child --uid ('name_error', '505bf71d ...)
+    ├─ python -m tractor._child --uid
+    │  ('spawner1', '7eab8462 ...')
+    │  └─ python -m tractor._child --uid
+    │     ('spawn_until_0', '3720602b ...')
+    │     └─ python -m tractor._child --uid
+    │        ('name_error', '505bf71d ...')
     │
-    └─ python -m tractor._child --uid ('spawner0', '1d42012b ...)
-       └─ python -m tractor._child --uid ('name_error', '6c2733b8 ...)
+    └─ python -m tractor._child --uid
+       ('spawner0', '1d42012b ...')
+       └─ python -m tractor._child --uid
+          ('name_error', '6c2733b8 ...')
 
     '''
     async with (

examples/debugging/root_self_cancelled_w_error.py

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

examples/debugging/root_timeout_while_child_crashed.py

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

examples/debugging/shielded_pause.py

@@ -3,8 +3,15 @@ import tractor
 
 
 async def cancellable_pause_loop(
-    task_status: trio.TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED
-):
+    task_status: trio.TaskStatus[
+        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):
@@ -30,7 +37,11 @@ async def cancellable_pause_loop(
             await trio.lowlevel.checkpoint()
 
 
-async def pm_on_cancelled():
+async def pm_on_cancelled() -> None:
+    '''
+    Compare shielded and unshielded post-mortem entry.
+
+    '''
     async with trio.open_nursery() as tn:
         tn.cancel_scope.cancel()
         try:
@@ -56,7 +67,7 @@ async def pm_on_cancelled():
 
 
 async def cancelled_before_pause(
-):
+) -> None:
     '''
     Verify that using a shielded pause works despite surrounding
     cancellation called state in the calling task.
@@ -71,7 +82,11 @@ async def cancelled_before_pause(
     await pm_on_cancelled()
 
 
-async def main():
+async def main() -> None:
+    '''
+    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,10 +1,15 @@
 import platform
+from collections.abc import AsyncIterator
 
 import tractor
 import trio
 
 
-async def gen():
+async def gen() -> AsyncIterator[str]:
+    '''
+    Yield values around debugger pauses.
+
+    '''
     yield 'yo'
     await tractor.pause()
     yield 'yo'
@@ -15,11 +20,15 @@ async def gen():
 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)
 
@@ -34,15 +43,18 @@ async def just_bp(
 
 
 
-async def main():
+async def main() -> None:
+    '''
+    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 = 'uds'
+        tpt: str = 'uds'
     else:
         # XXX, precisely we can't use pytest's tmp-path generation
         # for tests.. apparently because:
@@ -59,7 +71,7 @@ async def main():
         enable_transports=[tpt],
         loglevel='devx',
     ) as an:
-        p = await an.start_actor(
+        p: tractor.Portal = 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():
+async def breakpoint_forever() -> None:
     '''
     Indefinitely re-enter debugger in child actor.
 
@@ -12,8 +12,11 @@ async def breakpoint_forever():
         await tractor.pause()
 
 
-async def main():
+async def main() -> None:
+    '''
+    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,11 +2,19 @@ import trio
 import tractor
 
 
-async def name_error():
+async def name_error() -> None:
+    '''
+    Deliberately raise a ``NameError`` in a subactor.
+
+    '''
     getattr(doggypants)  # noqa (on purpose)
 
 
-async def main():
+async def main() -> None:
+    '''
+    Surface a subactor `NameError` at the waiting root task.
+
+    '''
     async with tractor.open_nursery(
         debug_mode=True,
     ) as an:

examples/discovery/dedicated_registrar.py

@@ -0,0 +1,394 @@
+'''
+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,4 +1,6 @@
 import time
+from typing import AsyncIterator
+
 import trio
 import tractor
 from tractor import (
@@ -9,14 +11,19 @@ from tractor import (
 
 
 # this is the first 2 actors, streamer_1 and streamer_2
-async def stream_data(seed):
+async def stream_data(seed: int) -> AsyncIterator[int]:
+    '''
+    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):
+async def aggregate(seed: int) -> AsyncIterator[int]:
     '''
     Ensure that the two streams we receive match but only stream
     a single set of values to the parent.
@@ -25,30 +32,47 @@ async def aggregate(seed):
     an: ActorNursery
     async with tractor.open_nursery() as an:
         portals: list[Portal] = []
+        i: int
         for i in range(1, 3):
 
             # fork/spawn call
-            portal = await an.start_actor(
+            portal: 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(portal, send_chan):
+        async def push_to_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:
-                async with portal.open_stream_from(stream_data, seed=seed) as stream:
+                stream: MsgStream
+                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)
 
-            print(f"FINISHED ITERATING {portal.channel.uid}")
+            uid: tuple[str, str] = portal.chan.uid
+            print(f'FINISHED ITERATING {uid}')
 
-        # spawn 2 trio tasks to collect streams and push to a local queue
+        # spawn 2 trio tasks to collect streams and push to a local
+        # queue
+        n: trio.Nursery
         async with trio.open_nursery() as n:
 
             for portal in portals:
@@ -61,8 +85,9 @@ async def aggregate(seed):
             # close this local task's reference to send side
             await send_chan.aclose()
 
-            unique_vals = set()
+            unique_vals: set[int] = set()
             async with recv_chan:
+                value: int
                 async for value in recv_chan:
                     if value not in unique_vals:
                         unique_vals.add(value)
@@ -71,11 +96,11 @@ async def aggregate(seed):
 
                 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("AGGREGATOR COMPLETE!")
+        print('WAITING on `ActorNursery` to finish')
+    print('AGGREGATOR COMPLETE!')
 
 
 async def main() -> list[int]:
@@ -94,8 +119,8 @@ async def main() -> list[int]:
         # debug_mode=True,
     ) as an:
 
-        seed = int(1e3)
-        pre_start = time.time()
+        seed: int = int(1e3)
+        pre_start: float = time.time()
 
         portal: Portal = await an.start_actor(
             name='aggregator',
@@ -108,23 +133,27 @@ async def main() -> list[int]:
             seed=seed,
         ) as stream:
 
-            start = time.time()
+            start: float = time.time()
             # the portal call returns exactly what you'd expect
-            # as if the remote "aggregate" function was called locally
+            # as if the remote "aggregate" function was called
+            # 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 = {time.time() - start}\n"
-            f"STREAM + SPAWN TIME = {time.time() - pre_start}\n"
+            f'STREAM TIME = {stream_time}\n'
+            f'STREAM + SPAWN TIME = {total_time}\n'
         )
         assert result_stream == list(range(seed))
         return result_stream
 
 
 if __name__ == '__main__':
-    final_stream = trio.run(main)
+    final_stream: list[int] = trio.run(main)

examples/infected_asyncio_echo_server.py

@@ -13,7 +13,10 @@ 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():``
@@ -28,9 +31,15 @@ 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):
@@ -38,39 +47,49 @@ 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 = await chan.receive()
+                out: int = await chan.receive()
                 # echo back to parent actor-task
                 await stream.send(out)
 
 
-async def main():
+async def main() -> None:
+    '''
+    Run the infected asyncio echo-server example.
 
+    '''
+    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        p = await an.start_actor(
+        portal: tractor.Portal = await an.start_actor(
             'aio_server',
             enable_modules=[__name__],
             infect_asyncio=True,
         )
-        async with p.open_context(
+        ctx: tractor.Context
+        first: str
+        async with portal.open_context(
             trio_to_aio_echo_server,
         ) as (ctx, first):
 
             assert first == 'start'
 
-            count = 0
+            count: int = 0
+            stream: tractor.MsgStream
             async with ctx.open_stream() as stream:
 
-                delays = []
-                send = time.time()
+                delays: list[float] = []
+                send: float = time.time()
 
                 await stream.send(count)
+                msg: int
                 async for msg in stream:
-                    recv = time.time()
+                    recv: float = time.time()
                     delays.append(recv - send)
                     assert msg == count
                     count += 1
@@ -81,7 +100,7 @@ async def main():
                         break
 
         print(f'mean round trip rate (Hz): {1/mean(delays)}')
-        await p.cancel_actor()
+        await portal.cancel_actor()
 
 
 if __name__ == '__main__':

examples/integration/mpi4py/inherit_parent_main.py

@@ -1,9 +1,10 @@
-"""
+'''
 Integration test: spawning tractor actors from an MPI process.
 
-When a parent is launched via ``mpirun``, Open MPI sets ``OMPI_*`` env
-vars that bind ``MPI_Init`` to the ``orted`` daemon.  Tractor children
-inherit those env vars, so if ``inherit_parent_main=True`` (the default)
+When a parent is launched via ``mpirun``, Open MPI sets
+``OMPI_*`` env vars that bind ``MPI_Init`` to the ``orted``
+daemon. Tractor children inherit those env vars, so if
+``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``::
@@ -12,13 +13,15 @@ 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 entirely.
+separate importable module (``_child``) avoids the re-import
+entirely.
 
 Usage::
 
     mpirun --allow-run-as-root -np 1 python -m \
         examples.integration.mpi4py.inherit_parent_main
-"""
+
+'''
 
 from mpi4py import MPI
 
@@ -30,21 +33,27 @@ from ._child import child_fn
 
 
 async def main() -> None:
-    rank = MPI.COMM_WORLD.Get_rank()
-    print(f"[parent] rank={rank}  pid={os.getpid()}", flush=True)
+    '''
+    Spawn an MPI-safe child without replaying the parent main.
 
+    '''
+    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 = await an.start_actor(
+        portal: tractor.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 = await portal.run(child_fn)
-        print(f"[parent] got: {result}", flush=True)
+        result: str = await portal.run(child_fn)
+        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,3 +1,5 @@
+from typing import AsyncIterator
+
 import trio
 import tractor
 
@@ -5,17 +7,30 @@ import tractor
 log = tractor.log.get_logger('multiportal')
 
 
-async def stream_data(seed=10):
-    log.info("Starting stream task")
+async def stream_data(seed: int = 10) -> AsyncIterator[int]:
+    '''
+    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(p, consumed):
+async def stream_from_portal(
+    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)
@@ -23,20 +38,33 @@ async def stream_from_portal(p, consumed):
                 consumed.append(item)
 
 
-async def main():
+async def main() -> None:
+    '''
+    Consume two concurrent streams through one portal.
 
+    '''
+    an: tractor.ActorNursery
     async with tractor.open_nursery(loglevel='info') as an:
 
-        p = await an.start_actor('stream_boi', enable_modules=[__name__])
+        portal: tractor.Portal = await an.start_actor(
+            'stream_boi',
+            enable_modules=[__name__],
+        )
 
-        consumed = []
+        consumed: list[int] = []
 
+        n: trio.Nursery
         async with trio.open_nursery() as n:
-            for i in range(2):
-                n.start_soon(stream_from_portal, p, consumed)
+            for _ in range(2):
+                n.start_soon(
+                    stream_from_portal,
+                    portal,
+                    consumed,
+                )
 
-        # both streaming consumer tasks have completed and so we should
-        # have nothing in our list thanks to single threadedness
+        # both streaming consumer tasks have completed and so we
+        # should have nothing in our list thanks to single
+        # threadedness
         assert not consumed
 
         await an.cancel()

examples/nested_actor_tree.py

@@ -41,6 +41,7 @@ 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):
@@ -52,21 +53,28 @@ async def fan_out_squares(
             )
         # unblock the parent's `.open_context()` entry and
         # report which leaves came up.
-        await ctx.started(
-            [p.chan.aid.name for p in portals]
-        )
+        leaf_names: list[str] = [
+            portal.chan.aid.name
+            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(
@@ -83,11 +91,18 @@ 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 = await an.start_actor(
+        portal: tractor.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,18 +1,23 @@
-"""
+'''
 Demonstration of the prime number detector example from the
 ``concurrent.futures`` docs:
 
-https://docs.python.org/3/library/concurrent.futures.html#processpoolexecutor-example
+https://docs.python.org/3/library/concurrent.futures.html\
+#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 Callable
+from typing import (
+    AsyncIterator,
+    Awaitable,
+    Callable,
+)
 import itertools
 import math
 import time
@@ -21,7 +26,12 @@ import tractor
 import trio
 
 
-PRIMES = [
+type ActorMap = Callable[
+    [Callable[[int], Awaitable[bool]], list[int]],
+    AsyncIterator[tuple[int, bool]],
+]
+
+PRIMES: list[int] = [
     112272535095293,
     112582705942171,
     112272535095293,
@@ -31,7 +41,11 @@ PRIMES = [
 ]
 
 
-async def is_prime(n):
+async def is_prime(n: int) -> bool:
+    '''
+    Return whether ``n`` is prime.
+
+    '''
     if n < 2:
         return False
     if n == 2:
@@ -47,23 +61,32 @@ async def is_prime(n):
 
 
 @acm
-async def worker_pool(workers=4):
-    """Though it's a trivial special case for ``tractor``, the well
+async def worker_pool(
+    workers: int = 4,
+) -> 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 = []
+        portals: list[tractor.Portal] = []
+        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 runtime) and
-            # stores it's "portal" for later use to "submit jobs" (ugh).
+            # this starts a new sub-actor (process + trio
+            # runtime) and stores it's "portal" for later use to
+            # "submit jobs" (ugh).
             portals.append(
                 await an.start_actor(
                     f'worker_{i}',
@@ -72,17 +95,36 @@ async def worker_pool(workers=4):
             )
 
         async def _map(
-            worker_func: Callable[[int], bool],
-            sequence: list[int]
-        ) -> list[bool]:
+            worker_func: Callable[[int], Awaitable[bool]],
+            sequence: list[int],
+        ) -> AsyncIterator[tuple[int, bool]]:
+            '''
+            Dispatch values across workers and yield their results.
 
-            # define an async (local) task to collect results from workers
-            async def send_result(func, value, portal):
-                await snd_chan.send((value, await portal.run(func, n=value)))
+            '''
+            # define an async (local) task to collect results from
+            # workers
+            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:
 
-                for value, portal in zip(sequence, itertools.cycle(portals)):
+                value: int
+                portal: tractor.Portal
+                for value, portal in zip(
+                    sequence,
+                    itertools.cycle(portals),
+                ):
                     tn.start_soon(
                         send_result,
                         worker_func,
@@ -101,21 +143,30 @@ async def worker_pool(workers=4):
         await an.cancel()
 
 
-async def main():
+async def main() -> None:
+    '''
+    Report primality results from a pool of actors.
 
+    '''
+    actor_map: ActorMap
     async with worker_pool() as actor_map:
 
-        start = time.time()
+        start: float = 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}')
 
-        print(f'processing took {time.time() - start} seconds')
+        elapsed: float = time.time() - start
+        print(f'processing took {elapsed} seconds')
 
 
 if __name__ == '__main__':
-    start = time.time()
+    start: float = time.time()
     trio.run(main)
-    print(f'script took {time.time() - start} seconds')
+    elapsed: float = time.time() - start
+    print(f'script took {elapsed} seconds')

examples/parallelism/single_func.py

@@ -1,20 +1,23 @@
-"""
+'''
 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():
+async def burn_cpu() -> int:
+    '''
+    Burn CPU briefly and return the current process ID.
 
-    pid = os.getpid()
+    '''
+    pid: int = os.getpid()
 
     # burn a core @ ~ 50kHz
     for _ in range(50000):
@@ -23,8 +26,11 @@ async def burn_cpu():
     return pid
 
 
-async def main():
+async def main() -> None:
+    '''
+    Run ``burn_cpu()`` in the parent and a subactor.
 
+    '''
     async with trio.open_nursery() as tn:
 
         # burn rubber in the parent too
@@ -32,9 +38,9 @@ async def main():
 
         # run the same func as the lone task in a subactor,
         # block on and collect its PID as the caller-side result
-        pid = await tractor.to_actor.run(burn_cpu)
+        pid: int = 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,20 +1,24 @@
-
 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():
+async def main() -> None:
     '''
     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 (
@@ -27,7 +31,9 @@ async def main():
         trio.open_nursery() as tn,
     ):
 
-        for (name, portal) in portal_map.items():
+        name: str
+        portal: tractor.Portal
+        for name, portal in portal_map.items():
             tn.start_soon(
                 portal.run,
                 sleepy_jane,

examples/remote_error_propagation.py

@@ -2,18 +2,30 @@ import trio
 import tractor
 
 
-async def assert_err():
+async def assert_err() -> None:
+    '''
+    Raise an assertion error in the current actor.
+
+    '''
     assert 0
 
 
-async def main():
+async def main() -> None:
+    '''
+    Propagate a failing one-shot task from a subactor.
+
+    '''
+    an: tractor.ActorNursery
     async with tractor.open_nursery() as an:
-        real_actors = []
+        real_actors: list[tractor.Portal] = []
+        i: int
         for i in range(3):
-            real_actors.append(await an.start_actor(
+            real_actors.append(
+                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..
@@ -28,4 +40,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,16 +9,19 @@ 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 = 0
+        count: int = 0
+        msg: str
         async for msg in stream:
 
             assert msg == 'ping'
@@ -30,15 +33,22 @@ 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 = await an.start_actor(
+        portal: tractor.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(
@@ -52,10 +62,11 @@ async def main() -> None:
 
             assert sent == 11
 
-            count = 0
+            count: int = 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,11 +35,13 @@ async def client_task() -> None:
 
     '''
     # a lookup miss yields `None` (not an error).
-    async with tractor.find_actor('no_such_svc') as portal:
-        assert portal is None
+    maybe_portal: tractor.Portal|None
+    async with tractor.find_actor('no_such_svc') as maybe_portal:
+        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,
@@ -49,13 +51,19 @@ 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 = await an.start_actor(
+        portal: tractor.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,19 +1,29 @@
 import trio
 import tractor
 
-tractor.log.get_console_log("INFO")
+tractor.log.get_console_log('INFO')
 
 
-async def main(service_name):
+async def main(service_name: str) -> None:
+    '''
+    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 portal:
-            print(f"Registrar is listening on {portal.channel}")
+        async with tractor.get_registry() as reg_portal:
+            print(
+                f'Registrar is listening on {reg_portal.channel}'
+            )
 
-        async with tractor.wait_for_actor(service_name) as sockaddr:
-            print(f"my_service is found at {sockaddr}")
+        actor_portal: tractor.Portal
+        async with tractor.wait_for_actor(
+            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,6 +28,7 @@ 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;
@@ -37,15 +38,17 @@ async def tick_stream(
 async def consume(
     name: str,
     stream: tractor.MsgStream,
-    task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED,
+    task_status: trio.TaskStatus[None] = 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)
@@ -54,11 +57,19 @@ 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 = await an.start_actor(
+        portal: tractor.Portal = await an.start_actor(
             'ticker',
             enable_modules=[__name__],
         )
+        ctx: tractor.Context
+        first: int
+        stream: tractor.MsgStream
         async with (
             portal.open_context(
                 tick_stream,
@@ -67,9 +78,11 @@ 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,3 +1,4 @@
+from collections.abc import AsyncIterator
 from contextlib import (
     asynccontextmanager as acm,
 )
@@ -16,7 +17,11 @@ _lock: trio.Lock|None = None
 
 @acm
 async def acquire_singleton_lock(
-) -> None:
+) -> AsyncIterator[trio.Lock]:
+    '''
+    Acquire and yield the process-wide lock.
+
+    '''
     global _lock
     if _lock is None:
         log.info('Allocating LOCK')
@@ -32,8 +37,15 @@ async def acquire_singleton_lock(
 
 
 async def hold_lock_forever(
-    task_status=trio.TASK_STATUS_IGNORED
-):
+    task_status: trio.TaskStatus[
+        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,
@@ -46,7 +58,12 @@ 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,
 
@@ -58,7 +75,7 @@ async def main(
             from tractor.trionics import _taskc
             _taskc._mask_cases.clear()
 
-        _lock = await tn.start(
+        _held_lock: trio.Lock = await tn.start(
             hold_lock_forever,
         )
         with trio.move_on_after(0.2):
@@ -74,8 +91,8 @@ if __name__ == '__main__':
     tractor.log.get_console_log(level='info')
     for case in [True, False]:
         log.info(
-            f'\n'
-            f'------ RUNNING SCRIPT TRIAL ------\n'
+            '\n'
+            '------ RUNNING SCRIPT TRIAL ------\n'
             f'ignore_special_cases: {case!r}\n'
         )
         trio.run(partial(

examples/trio/send_chan_aclose_masks_beg.py

@@ -1,3 +1,4 @@
+from collections.abc import Iterator
 from contextlib import (
     contextmanager as cm,
     # TODO, any diff in async case(s)??
@@ -17,7 +18,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
@@ -30,7 +31,7 @@ def teardown_on_exc(
     except BaseException as _berr:
         berr = _berr
         log.exception(
-            f'Handling termination teardown in child due to,\n'
+            'Handling termination teardown in child due to,\n'
             f'{berr!r}\n'
         )
         if raise_from_handler:
@@ -54,14 +55,18 @@ def teardown_on_exc(
 
 
 async def finite_stream_to_rent(
-    tx: trio.abc.SendChannel,
+    tx: trio.abc.SendChannel[int],
     child_errors_mid_stream: bool,
     raise_unmasked: bool,
 
     task_status: trio.TaskStatus[
-        trio.CancelScope,
+        trio.CancelScope|None,
     ] = 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()`
@@ -134,19 +139,26 @@ 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 = await tn.start(
+        _child_cs: trio.CancelScope|None = await tn.start(
             partial(
                 finite_stream_to_rent,
                 child_errors_mid_stream=child_errors_mid_stream,
@@ -154,6 +166,7 @@ async def main(
                 tx=tx,
             )
         )
+        msg: int
         async for msg in rx:
             log.debug(
                 f'Rent rx {msg!r}\n'
@@ -162,12 +175,13 @@ async def main(
             # simulate some external cancellation
             # request **JUST BEFORE** the child errors.
             if msg == 65:
-                log.cancel(
-                    f'Cancelling parent on,\n'
-                    f'msg={msg}\n'
-                    f'\n'
-                    f'Simulates OOB cancel request!\n'
-                )
+                cancel_msg: str = (
+                    'Cancelling parent on,\n'
+                    'msg={msg}\n'
+                    '\n'
+                    'Simulates OOB cancel request!\n'
+                ).format(msg=msg)
+                log.cancel(cancel_msg)
                 tn.cancel_scope.cancel()
 
 
@@ -176,8 +190,8 @@ if __name__ == '__main__':
     tractor.log.get_console_log(level='info')
     for case in [True, False]:
         log.info(
-            f'\n'
-            f'------ RUNNING SCRIPT TRIAL ------\n'
+            '\n'
+            '------ RUNNING SCRIPT TRIAL ------\n'
             f'child_errors_midstream: {case!r}\n'
         )
         try:

examples/typed_payloads.py

@@ -46,7 +46,9 @@ 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
@@ -59,11 +61,19 @@ 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 = await an.start_actor(
+        portal: tractor.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,
@@ -72,6 +82,7 @@ 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,17 +21,22 @@ async def report_addr() -> str:
     Return this actor's own accept (bind) addr + pid.
 
     '''
-    actor = tractor.current_actor()
-    addr: tuple = actor.accept_addr
+    actor: tractor.Actor = tractor.current_actor()
+    addr: tuple[str, int|str] = 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 = await an.start_actor(
+        portal: tractor.Portal = await an.start_actor(
             'uds_child',
             enable_modules=[__name__],
         )
@@ -51,7 +56,8 @@ async def main() -> None:
         )
         # ask the child for its OWN distinct bind addr: another
         # socket-file path under the runtime dir.
-        print(f'child says: {await portal.run(report_addr)}')
+        child_report: str = await portal.run(report_addr)
+        print(f'child says: {child_report}')
         await portal.cancel_actor()