From a279cb39a950415ff62e8632d7fe5447087f618d Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:32:49 -0400 Subject: [PATCH 01/12] Expand caps-based-msging docs w/ #365, #376 + tests The "Toward capability-based msging" section only pointed at the `#196`/`#36` epics. Fold in the concrete recent state, - `#365` as the most recent step: driving the whole `pld_spec` off plain type-annotations (e.g. annotating a context's `open_stream()` with `msgspec.Struct` subtypes) rather than explicit `pld_spec=` kwargs. - clarify that the decorator-level `@tractor.context(pld_spec=...)` is already the higher-level path (vs the lower-level `tractor.msg._ops.limit_plds()` escape hatch), pointing at `tests/msg/test_pldrx_limiting.py` + `test_ext_types_msgspec.py` which exercise both. - `#376` (from @guilledk, `auto_codecs` branch) as the drafted public factory API for the `enc_hook`/`dec_hook` pair (today only reachable via `tractor.msg._ops`). Addresses the caps-based-msging bullet in #472. (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- docs/guide/msging.rst | 30 +++++++++++++++++++++++++++--- 1 file changed, 27 insertions(+), 3 deletions(-) diff --git a/docs/guide/msging.rst b/docs/guide/msging.rst index 5c5a4d12..eb5f8b8d 100644 --- a/docs/guide/msging.rst +++ b/docs/guide/msging.rst @@ -238,9 +238,31 @@ 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``. + +On the codec-hook side, the ``enc_hook``/``dec_hook`` pair is +today only reachable via ``tractor.msg._ops``; a public *factory* +API for them is drafted in `#376`_ (from +`@guilledk `_, on the +`auto_codecs `_ +branch) — the likely 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 +280,5 @@ 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 From 4c507ea13c98978aea6d42b20fa86e4930892d69 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:32:49 -0400 Subject: [PATCH 02/12] Add a dedicated-registrar example + discovery guide Add a runnable `examples/dedicated_registrar.py` + a "A dedicated registrar" subsection in `guide/discovery.rst` demoing the registrar decoupled from any app tree's root: boot a bare `tractor.run_daemon([], registry_addrs=[...])` as its own process (a root actor that does nothing but hold the registry), point the app tree at the same `registry_addrs`, and discover a service *through* that external registrar. This is the buildable-today form of the #472 "Registrar-as-subsystem (not the root actor)" bullet. Two constraints are called out inline as follow-ups: `enable_transports` is single-proto per runtime (no multi-backend registrar yet), and a registrar can only be a root (no `actor_cls` hook on `start_actor()` to spawn one as a subactor). (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- docs/guide/discovery.rst | 22 ++++++ examples/dedicated_registrar.py | 122 ++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+) create mode 100644 examples/dedicated_registrar.py diff --git a/docs/guide/discovery.rst b/docs/guide/discovery.rst index ea993b52..4ce2d6ad 100644 --- a/docs/guide/discovery.rst +++ b/docs/guide/discovery.rst @@ -65,6 +65,28 @@ 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. Boot +a bare ``tractor.run_daemon([], registry_addrs=[...])`` (a root +actor that does nothing but hold the registry), point your app +tree at the same ``registry_addrs``, and every actor discovers +through that *external* registrar instead of a tree-local one: + +.. literalinclude:: ../../examples/dedicated_registrar.py + :caption: examples/dedicated_registrar.py + :language: python + +This is the "registrar as a subsystem, not the 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 ----------------- diff --git a/examples/dedicated_registrar.py b/examples/dedicated_registrar.py new file mode 100644 index 00000000..d80af49b --- /dev/null +++ b/examples/dedicated_registrar.py @@ -0,0 +1,122 @@ +''' +Run a *dedicated* registrar as its own standalone process — decoupled +from your app's root actor — and discover a service *through* it. + +Normally the registrar **is** the root actor of your tree. Here we +instead boot a separate `tractor.run_daemon([], registry_addrs=[...])` +process whose *sole* job is to be the registry, then point our app +tree at it via `registry_addrs`. Because a registrar is already +reachable at that addr, our app's root actor does NOT become one — it +registers with (and discovers through) the external daemon. That's the +"registrar as a subsystem, not the root actor" pattern. + +NB: `enable_transports` is single-proto per-runtime today (see +`tractor._root`), so this demos one transport; a genuinely +multi-backend registrar (and spawning one as a *sub*actor of a shared +tree) are future runtime work — see the #472 follow-ups. + +''' +from contextlib import suppress +import signal +import socket +import subprocess +import sys +import time + +import trio +import tractor + + +# the fixed addr the dedicated registrar binds and everyone points at. +REG_ADDR: tuple[str, int] = ('127.0.0.1', 1717) + + +def _wait_registrar_ready( + addr: tuple[str, int], + proc: subprocess.Popen, + deadline: float = 10.0, +) -> None: + ''' + Active-poll the registrar's bind addr until it accepts a + connection (proving it's booted + listening), bailing early if + the daemon proc dies during startup. + + ''' + end: float = time.monotonic() + deadline + while time.monotonic() < end: + if proc.poll() is not None: + raise RuntimeError( + f'registrar died on startup (rc={proc.returncode})' + ) + with suppress(OSError): + with socket.create_connection(addr, timeout=0.1): + return + time.sleep(0.05) + raise TimeoutError(f'registrar never came up @ {addr}') + + +async def greet() -> str: + '''A trivial service task any peer can RPC by name.''' + return f'hello from {tractor.current_actor().name}!' + + +async def app() -> None: + ''' + Point our app tree at the EXTERNAL registrar (not its own root) + via `registry_addrs`, register a named service, then discover + + RPC it purely by name. + + ''' + an: tractor.ActorNursery + async with tractor.open_nursery( + registry_addrs=[REG_ADDR], + enable_transports=['tcp'], + ) as an: + # this subactor registers with the DEDICATED registrar @ + # REG_ADDR (our root is a plain peer, not the registry). + await an.start_actor( + 'greeter', + enable_modules=[__name__], + ) + # discover it *through the external registrar*, by name only. + portal: tractor.Portal + async with tractor.wait_for_actor('greeter') as portal: + print(f'found `greeter` via dedicated registrar @ {REG_ADDR}') + print(await portal.run(greet)) + await an.cancel() + + +def main() -> None: + # boot the dedicated registrar as its own process/tree: an empty + # `enable_modules` `run_daemon()` is just a root actor that does + # nothing but hold + serve the registry. + code: str = ( + 'import tractor; ' + f'tractor.run_daemon([], registry_addrs={[REG_ADDR]!r}, ' + "enable_transports=['tcp'], loglevel='error')" + ) + registrar: subprocess.Popen = subprocess.Popen( + [sys.executable, '-c', code], + # the registry is a quiet background service; hush its logs + + # expected SIGINT-teardown traceback so the demo output stays + # focused on the discovery flow. + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + try: + _wait_registrar_ready(REG_ADDR, registrar) + print( + f'dedicated registrar up @ {REG_ADDR} ' + f'(pid {registrar.pid})' + ) + trio.run(app) + finally: + # graceful SIGINT teardown of the standalone registrar. + registrar.send_signal(signal.SIGINT) + with suppress(subprocess.TimeoutExpired): + registrar.wait(timeout=10) + print('dedicated registrar shut down') + + +if __name__ == '__main__': + main() From a6c1158853179dd36f1208d3238f33fbd9ee0b19 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:32:49 -0400 Subject: [PATCH 03/12] Add basic typing to the `debugging/` examples Sweep the `examples/debugging/` set for basic typing: add `-> None` to all 16 bare `async def main()`s and annotate the clean single-line `open_nursery()` bindings as `tractor.ActorNursery`. Kept to the unambiguous, runtime-safe cases (these breakpoint/crash demos can't be run headless); the heterogeneous multi-line/paren-group nursery bindings + `current_actor()` returns are left for a later pass. Continues the examples-typing bullet in (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- examples/debugging/fast_error_in_root_after_spawn.py | 2 +- examples/debugging/multi_daemon_subactors.py | 2 +- .../multi_nested_subactors_error_up_through_nurseries.py | 2 +- examples/debugging/multi_subactor_root_errors.py | 2 +- examples/debugging/multi_subactors.py | 2 +- examples/debugging/per_actor_debug.py | 2 +- examples/debugging/pm_in_subactor.py | 2 +- examples/debugging/root_actor_breakpoint.py | 2 +- examples/debugging/root_actor_error.py | 2 +- examples/debugging/root_cancelled_but_child_is_in_tty_lock.py | 2 +- examples/debugging/root_self_cancelled_w_error.py | 2 +- examples/debugging/root_timeout_while_child_crashed.py | 2 +- examples/debugging/shielded_pause.py | 2 +- examples/debugging/subactor_bp_in_ctx.py | 2 +- examples/debugging/subactor_breakpoint.py | 2 +- examples/debugging/subactor_error.py | 2 +- 16 files changed, 16 insertions(+), 16 deletions(-) diff --git a/examples/debugging/fast_error_in_root_after_spawn.py b/examples/debugging/fast_error_in_root_after_spawn.py index 5c2fdce2..a3953d36 100644 --- a/examples/debugging/fast_error_in_root_after_spawn.py +++ b/examples/debugging/fast_error_in_root_after_spawn.py @@ -35,7 +35,7 @@ async def open_ctx( assert first is None -async def main(): +async def main() -> None: async with tractor.open_nursery( debug_mode=True, diff --git a/examples/debugging/multi_daemon_subactors.py b/examples/debugging/multi_daemon_subactors.py index e313803a..95822f93 100644 --- a/examples/debugging/multi_daemon_subactors.py +++ b/examples/debugging/multi_daemon_subactors.py @@ -20,7 +20,7 @@ async def name_error(): getattr(doggypants) # noqa -async def main(): +async def main() -> None: ''' Test breakpoint in a streaming actor. diff --git a/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py b/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py index 9929b498..2895f0e6 100644 --- a/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py +++ b/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py @@ -63,7 +63,7 @@ async def spawn_until(depth=0): # TODO: notes on the new boxed-relayed errors through proxy actors -async def main(): +async def main() -> None: """The main ``tractor`` routine. The process tree should look as approximately as follows when the debugger diff --git a/examples/debugging/multi_subactor_root_errors.py b/examples/debugging/multi_subactor_root_errors.py index 5aa3a4ff..b934c515 100644 --- a/examples/debugging/multi_subactor_root_errors.py +++ b/examples/debugging/multi_subactor_root_errors.py @@ -23,7 +23,7 @@ async def spawn_error(): ) -async def main(): +async def main() -> None: """The main ``tractor`` routine. The process tree should look as approximately as follows: diff --git a/examples/debugging/multi_subactors.py b/examples/debugging/multi_subactors.py index 63ab5404..f2ea18c9 100644 --- a/examples/debugging/multi_subactors.py +++ b/examples/debugging/multi_subactors.py @@ -25,7 +25,7 @@ async def spawn_error(): ) -async def main(): +async def main() -> None: """The main ``tractor`` routine. The process tree should look as approximately as follows: diff --git a/examples/debugging/per_actor_debug.py b/examples/debugging/per_actor_debug.py index 189fb45e..c5abe450 100644 --- a/examples/debugging/per_actor_debug.py +++ b/examples/debugging/per_actor_debug.py @@ -5,7 +5,7 @@ async def die(): raise RuntimeError -async def main(): +async def main() -> None: async with tractor.open_nursery() as an: debug_actor = await an.start_actor( diff --git a/examples/debugging/pm_in_subactor.py b/examples/debugging/pm_in_subactor.py index a8f5048e..a9728a6b 100644 --- a/examples/debugging/pm_in_subactor.py +++ b/examples/debugging/pm_in_subactor.py @@ -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()`, diff --git a/examples/debugging/root_actor_breakpoint.py b/examples/debugging/root_actor_breakpoint.py index 55b4ca56..347123ef 100644 --- a/examples/debugging/root_actor_breakpoint.py +++ b/examples/debugging/root_actor_breakpoint.py @@ -2,7 +2,7 @@ import trio import tractor -async def main(): +async def main() -> None: async with tractor.open_root_actor( debug_mode=True, diff --git a/examples/debugging/root_actor_error.py b/examples/debugging/root_actor_error.py index fab46335..49359f16 100644 --- a/examples/debugging/root_actor_error.py +++ b/examples/debugging/root_actor_error.py @@ -2,7 +2,7 @@ import trio import tractor -async def main(): +async def main() -> None: async with tractor.open_root_actor( debug_mode=True, ): diff --git a/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py b/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py index 75e1c9a4..ca15530b 100644 --- a/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py +++ b/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py @@ -27,7 +27,7 @@ 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: diff --git a/examples/debugging/root_self_cancelled_w_error.py b/examples/debugging/root_self_cancelled_w_error.py index b3c15288..e5ebbacc 100644 --- a/examples/debugging/root_self_cancelled_w_error.py +++ b/examples/debugging/root_self_cancelled_w_error.py @@ -2,7 +2,7 @@ import trio import tractor -async def main(): +async def main() -> None: async with tractor.open_root_actor( debug_mode=True, loglevel='cancel', diff --git a/examples/debugging/root_timeout_while_child_crashed.py b/examples/debugging/root_timeout_while_child_crashed.py index 043cb5c7..11533a8e 100644 --- a/examples/debugging/root_timeout_while_child_crashed.py +++ b/examples/debugging/root_timeout_while_child_crashed.py @@ -7,7 +7,7 @@ async def key_error(): return {}['doggy'] -async def main(): +async def main() -> None: ''' Root is fail-after-cancelled while blocking and child RPC fails simultaneously. diff --git a/examples/debugging/shielded_pause.py b/examples/debugging/shielded_pause.py index 5a8c50e7..cfa8f4f8 100644 --- a/examples/debugging/shielded_pause.py +++ b/examples/debugging/shielded_pause.py @@ -71,7 +71,7 @@ async def cancelled_before_pause( await pm_on_cancelled() -async def main(): +async def main() -> None: async with tractor.open_nursery( debug_mode=True, ) as an: diff --git a/examples/debugging/subactor_bp_in_ctx.py b/examples/debugging/subactor_bp_in_ctx.py index 36d4b2b3..eafeb0b7 100644 --- a/examples/debugging/subactor_bp_in_ctx.py +++ b/examples/debugging/subactor_bp_in_ctx.py @@ -34,7 +34,7 @@ async def just_bp( -async def main(): +async def main() -> None: # !TODO, parametrize the --tpt-proto={key} with osenv vars just # like we do for loglevel/spawn-backend! diff --git a/examples/debugging/subactor_breakpoint.py b/examples/debugging/subactor_breakpoint.py index e3a4f250..0a047a21 100644 --- a/examples/debugging/subactor_breakpoint.py +++ b/examples/debugging/subactor_breakpoint.py @@ -12,7 +12,7 @@ async def breakpoint_forever(): await tractor.pause() -async def main(): +async def main() -> None: async with tractor.open_nursery( debug_mode=True, diff --git a/examples/debugging/subactor_error.py b/examples/debugging/subactor_error.py index 95c1fe12..fd280cb9 100644 --- a/examples/debugging/subactor_error.py +++ b/examples/debugging/subactor_error.py @@ -6,7 +6,7 @@ async def name_error(): getattr(doggypants) # noqa (on purpose) -async def main(): +async def main() -> None: async with tractor.open_nursery( debug_mode=True, ) as an: From 319868e92d0eeb14d51daafb268cd28b6aaef2ba Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:32:49 -0400 Subject: [PATCH 04/12] Type the docs-visible `examples/` scripts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Type the runtime objects (`ActorNursery`, `Portal`, `Context`, `trio.Nursery`) + fn signatures across the 16 highest-visibility, `literalinclude`-d `examples/` scripts, matching the front-page `we_are_processes.py` style — so the rendered guides show typed usage throughout, not just on the landing snippet. Spans the 3 quickstart-backing scripts + `single_func`, `remote_error_propagation`, `multiple_streams_one_portal`, `quick_cluster`, `service_discovery`, `service_daemon_discovery`, `asynchronous_generators`, `nested_actor_tree`, `concurrent_actors_primes`, `streaming_broadcast_fanout`, `rpc_bidir_streaming`, `infected_asyncio_echo_server`, `typed_payloads`. Annotation-only (no renames/logic changes); each runs green and the docs build stays warning-free. Part of the examples-typing bullet in #472. (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- examples/a_trynamic_first_scene.py | 7 ++++--- examples/actor_spawning_and_causality.py | 4 ++-- ...actor_spawning_and_causality_with_daemon.py | 6 +++--- examples/asynchronous_generators.py | 6 +++--- examples/full_fledged_streaming_service.py | 9 +++++---- examples/infected_asyncio_echo_server.py | 6 +++--- examples/multiple_streams_one_portal.py | 18 +++++++++++++----- examples/nested_actor_tree.py | 5 ++++- .../parallelism/concurrent_actors_primes.py | 15 ++++++++++----- examples/parallelism/single_func.py | 6 +++--- examples/quick_cluster.py | 3 ++- examples/remote_error_propagation.py | 6 +++--- examples/rpc_bidir_streaming.py | 2 +- examples/service_daemon_discovery.py | 4 +++- examples/service_discovery.py | 5 ++++- examples/streaming_broadcast_fanout.py | 4 +++- examples/typed_payloads.py | 3 ++- examples/uds_transport_actor_tree.py | 3 ++- 18 files changed, 70 insertions(+), 42 deletions(-) diff --git a/examples/a_trynamic_first_scene.py b/examples/a_trynamic_first_scene.py index 85eb23e7..0606e5f0 100644 --- a/examples/a_trynamic_first_scene.py +++ b/examples/a_trynamic_first_scene.py @@ -8,16 +8,17 @@ the_line = 'Hi my name is {}' tractor.log.get_console_log("INFO") -async def hi(): +async def hi() -> str: return the_line.format(tractor.current_actor().name) -async def say_hello(other_actor): +async def say_hello(other_actor: str) -> str: + portal: tractor.Portal async with tractor.wait_for_actor(other_actor) as portal: return await portal.run(hi) -async def main(): +async def main() -> None: """Main tractor entry point, the "master" process (for now acts as the "director"). """ diff --git a/examples/actor_spawning_and_causality.py b/examples/actor_spawning_and_causality.py index 00dc645c..e5a5790f 100644 --- a/examples/actor_spawning_and_causality.py +++ b/examples/actor_spawning_and_causality.py @@ -2,12 +2,12 @@ import trio import tractor -async def cellar_door(): +async def cellar_door() -> str: assert not tractor.is_root_process() return "Dang that's beautiful" -async def main(): +async def main() -> None: """The main ``tractor`` routine. """ # spawn a subactor, run ``cellar_door()`` as its lone task, diff --git a/examples/actor_spawning_and_causality_with_daemon.py b/examples/actor_spawning_and_causality_with_daemon.py index 2e6824c9..c75ab57f 100644 --- a/examples/actor_spawning_and_causality_with_daemon.py +++ b/examples/actor_spawning_and_causality_with_daemon.py @@ -2,19 +2,19 @@ import trio import tractor -async def movie_theatre_question(): +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(): +async def main() -> None: """The main ``tractor`` routine. """ 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__], diff --git a/examples/asynchronous_generators.py b/examples/asynchronous_generators.py index 237794a6..0202c98b 100644 --- a/examples/asynchronous_generators.py +++ b/examples/asynchronous_generators.py @@ -13,11 +13,11 @@ async def stream_forever() -> AsyncIterator[int]: await trio.sleep(0.01) -async def main(): +async def main() -> None: async with tractor.open_nursery() as an: - portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'donny', enable_modules=[__name__], ) @@ -25,7 +25,7 @@ async def main(): # this async for loop streams values from the above # async generator running in a separate process async with portal.open_stream_from(stream_forever) as stream: - count = 0 + count: int = 0 async for letter in stream: print(letter) count += 1 diff --git a/examples/full_fledged_streaming_service.py b/examples/full_fledged_streaming_service.py index 390a1b75..de4d5286 100644 --- a/examples/full_fledged_streaming_service.py +++ b/examples/full_fledged_streaming_service.py @@ -9,14 +9,14 @@ 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): 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): ''' Ensure that the two streams we receive match but only stream a single set of values to the parent. @@ -28,7 +28,7 @@ async def aggregate(seed): 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__], ) @@ -37,7 +37,7 @@ async def aggregate(seed): 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): # TODO: https://github.com/goodboy/tractor/issues/207 async with send_chan: @@ -49,6 +49,7 @@ async def aggregate(seed): print(f"FINISHED ITERATING {portal.channel.uid}") # 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: diff --git a/examples/infected_asyncio_echo_server.py b/examples/infected_asyncio_echo_server.py index e3ff2a09..e111459d 100644 --- a/examples/infected_asyncio_echo_server.py +++ b/examples/infected_asyncio_echo_server.py @@ -28,7 +28,7 @@ async def aio_echo_server( @tractor.context async def trio_to_aio_echo_server( ctx: tractor.Context, -): +) -> None: # this will block until the ``asyncio`` task sends a "first" # message. async with tractor.to_asyncio.open_channel_from( @@ -48,10 +48,10 @@ async def trio_to_aio_echo_server( await stream.send(out) -async def main(): +async def main() -> None: async with tractor.open_nursery() as an: - p = await an.start_actor( + p: tractor.Portal = await an.start_actor( 'aio_server', enable_modules=[__name__], infect_asyncio=True, diff --git a/examples/multiple_streams_one_portal.py b/examples/multiple_streams_one_portal.py index 3e592a45..45272b6c 100644 --- a/examples/multiple_streams_one_portal.py +++ b/examples/multiple_streams_one_portal.py @@ -5,7 +5,7 @@ import tractor log = tractor.log.get_logger('multiportal') -async def stream_data(seed=10): +async def stream_data(seed: int = 10): log.info("Starting stream task") for i in range(seed): @@ -13,7 +13,10 @@ async def stream_data(seed=10): await trio.sleep(0) # trigger scheduler -async def stream_from_portal(p, consumed): +async def stream_from_portal( + p: tractor.Portal, + consumed: list, +) -> None: async with p.open_stream_from(stream_data) as stream: async for item in stream: @@ -23,14 +26,19 @@ async def stream_from_portal(p, consumed): consumed.append(item) -async def main(): +async def main() -> None: + an: tractor.ActorNursery async with tractor.open_nursery(loglevel='info') as an: - p = await an.start_actor('stream_boi', enable_modules=[__name__]) + p: tractor.Portal = await an.start_actor( + 'stream_boi', + enable_modules=[__name__], + ) - consumed = [] + consumed: list = [] + n: trio.Nursery async with trio.open_nursery() as n: for i in range(2): n.start_soon(stream_from_portal, p, consumed) diff --git a/examples/nested_actor_tree.py b/examples/nested_actor_tree.py index a0c66b88..cb45b22c 100644 --- a/examples/nested_actor_tree.py +++ b/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): @@ -67,6 +68,7 @@ async def fan_out_squares( ) # 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,8 +85,9 @@ async def fan_out_squares( async def main() -> None: + 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__], ) diff --git a/examples/parallelism/concurrent_actors_primes.py b/examples/parallelism/concurrent_actors_primes.py index e5b32359..e38adeee 100644 --- a/examples/parallelism/concurrent_actors_primes.py +++ b/examples/parallelism/concurrent_actors_primes.py @@ -31,7 +31,7 @@ PRIMES = [ ] -async def is_prime(n): +async def is_prime(n: int) -> bool: if n < 2: return False if n == 2: @@ -47,7 +47,7 @@ async def is_prime(n): @acm -async def worker_pool(workers=4): +async def worker_pool(workers: int = 4): """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. @@ -57,7 +57,7 @@ async def worker_pool(workers=4): """ async with tractor.open_nursery() as an: - portals = [] + portals: list[tractor.Portal] = [] snd_chan, recv_chan = trio.open_memory_channel(len(PRIMES)) for i in range(workers): @@ -77,9 +77,14 @@ async def worker_pool(workers=4): ) -> list[bool]: # define an async (local) task to collect results from workers - async def send_result(func, value, portal): + async def send_result( + func: Callable, + value: int, + portal: tractor.Portal, + ): await snd_chan.send((value, await portal.run(func, n=value))) + tn: trio.Nursery async with trio.open_nursery() as tn: for value, portal in zip(sequence, itertools.cycle(portals)): @@ -101,7 +106,7 @@ async def worker_pool(workers=4): await an.cancel() -async def main(): +async def main() -> None: async with worker_pool() as actor_map: diff --git a/examples/parallelism/single_func.py b/examples/parallelism/single_func.py index 46fc4837..9513505d 100644 --- a/examples/parallelism/single_func.py +++ b/examples/parallelism/single_func.py @@ -12,7 +12,7 @@ import tractor import trio -async def burn_cpu(): +async def burn_cpu() -> int: pid = os.getpid() @@ -23,7 +23,7 @@ async def burn_cpu(): return pid -async def main(): +async def main() -> None: async with trio.open_nursery() as tn: @@ -32,7 +32,7 @@ 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}") diff --git a/examples/quick_cluster.py b/examples/quick_cluster.py index 3fa4ca2a..489912c1 100644 --- a/examples/quick_cluster.py +++ b/examples/quick_cluster.py @@ -9,12 +9,13 @@ async def sleepy_jane() -> None: 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 ( diff --git a/examples/remote_error_propagation.py b/examples/remote_error_propagation.py index db9eb9b5..405c626b 100644 --- a/examples/remote_error_propagation.py +++ b/examples/remote_error_propagation.py @@ -2,13 +2,13 @@ import trio import tractor -async def assert_err(): +async def assert_err() -> None: assert 0 -async def main(): +async def main() -> None: async with tractor.open_nursery() as an: - real_actors = [] + real_actors: list[tractor.Portal] = [] for i in range(3): real_actors.append(await an.start_actor( f'actor_{i}', diff --git a/examples/rpc_bidir_streaming.py b/examples/rpc_bidir_streaming.py index eb4f03cd..68c4f541 100644 --- a/examples/rpc_bidir_streaming.py +++ b/examples/rpc_bidir_streaming.py @@ -33,7 +33,7 @@ async def main() -> None: async with tractor.open_nursery() as an: - portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'rpc_server', enable_modules=[__name__], ) diff --git a/examples/service_daemon_discovery.py b/examples/service_daemon_discovery.py index a9086b8c..0abd4987 100644 --- a/examples/service_daemon_discovery.py +++ b/examples/service_daemon_discovery.py @@ -49,13 +49,15 @@ async def client_task() -> None: async def main() -> None: + 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. diff --git a/examples/service_discovery.py b/examples/service_discovery.py index 574ba019..ef678951 100644 --- a/examples/service_discovery.py +++ b/examples/service_discovery.py @@ -4,14 +4,17 @@ import tractor tractor.log.get_console_log("INFO") -async def main(service_name): +async def main(service_name: str) -> None: + an: tractor.ActorNursery async with tractor.open_nursery() as an: await an.start_actor(service_name) + portal: tractor.Portal async with tractor.get_registry() as portal: print(f"Registrar is listening on {portal.channel}") + sockaddr: tractor.Portal async with tractor.wait_for_actor(service_name) as sockaddr: print(f"my_service is found at {sockaddr}") diff --git a/examples/streaming_broadcast_fanout.py b/examples/streaming_broadcast_fanout.py index 9f6900ce..15708795 100644 --- a/examples/streaming_broadcast_fanout.py +++ b/examples/streaming_broadcast_fanout.py @@ -54,8 +54,9 @@ async def consume( async def main() -> None: + 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__], ) @@ -67,6 +68,7 @@ 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. diff --git a/examples/typed_payloads.py b/examples/typed_payloads.py index c6352de5..58acada6 100644 --- a/examples/typed_payloads.py +++ b/examples/typed_payloads.py @@ -59,8 +59,9 @@ async def point_doubler( async def main() -> None: + 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__], ) diff --git a/examples/uds_transport_actor_tree.py b/examples/uds_transport_actor_tree.py index 93a17626..6555329d 100644 --- a/examples/uds_transport_actor_tree.py +++ b/examples/uds_transport_actor_tree.py @@ -28,10 +28,11 @@ async def report_addr() -> str: async def main() -> None: + 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__], ) From 0be872ff97ecc4d51d63df69a85b48ca7470dfc3 Mon Sep 17 00:00:00 2001 From: goodboy Date: Thu, 2 Jul 2026 12:35:37 -0400 Subject: [PATCH 05/12] Type the remaining niche examples Finish the examples-typing sweep with the last non-docs-visible scripts: `-> None` on the two `trio/` behavior-demo mains (plus a `trio.TaskStatus` on `hold_lock_forever`) and nursery/portal typing on `integration/mpi4py/inherit_parent_main.py`. Leaves `concurrent_futures_primes` (a verbatim stdlib baseline) and `integration/open_context_and_sleep` (its tractor nursery is commented out) as-is, and the paren-group `trio.open_nursery()` bindings unannotated (no clean spot for a preceding annotation). Completes the examples-typing bullet in #472. (this patch was generated in some part by [`claude-code`][claude-code-gh]) [claude-code-gh]: https://github.com/anthropics/claude-code --- examples/integration/mpi4py/inherit_parent_main.py | 5 +++-- examples/trio/lockacquire_not_unmasked.py | 6 +++--- examples/trio/send_chan_aclose_masks_beg.py | 2 +- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/examples/integration/mpi4py/inherit_parent_main.py b/examples/integration/mpi4py/inherit_parent_main.py index 60e30a95..741d8a6b 100644 --- a/examples/integration/mpi4py/inherit_parent_main.py +++ b/examples/integration/mpi4py/inherit_parent_main.py @@ -33,15 +33,16 @@ async def main() -> None: rank = MPI.COMM_WORLD.Get_rank() print(f"[parent] rank={rank} pid={os.getpid()}", 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) + result: str = await portal.run(child_fn) print(f"[parent] got: {result}", flush=True) await portal.cancel_actor() diff --git a/examples/trio/lockacquire_not_unmasked.py b/examples/trio/lockacquire_not_unmasked.py index 2f979a00..19077706 100644 --- a/examples/trio/lockacquire_not_unmasked.py +++ b/examples/trio/lockacquire_not_unmasked.py @@ -32,8 +32,8 @@ async def acquire_singleton_lock( async def hold_lock_forever( - task_status=trio.TASK_STATUS_IGNORED -): + task_status: trio.TaskStatus = trio.TASK_STATUS_IGNORED, +) -> None: async with ( tractor.trionics.maybe_raise_from_masking_exc(), acquire_singleton_lock() as lock, @@ -46,7 +46,7 @@ async def main( ignore_special_cases: bool, loglevel: str = 'info', debug_mode: bool = True, -): +) -> None: async with ( trio.open_nursery() as tn, diff --git a/examples/trio/send_chan_aclose_masks_beg.py b/examples/trio/send_chan_aclose_masks_beg.py index e7f895b7..971d8678 100644 --- a/examples/trio/send_chan_aclose_masks_beg.py +++ b/examples/trio/send_chan_aclose_masks_beg.py @@ -134,7 +134,7 @@ async def main( raise_unmasked: bool = False, loglevel: str = 'info', -): +) -> None: tractor.log.get_console_log(level=loglevel) # the `.aclose()` being checkpoints on these From 96a18382100ca63cca1baa84ba8ed8102309c488 Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 18:58:43 -0400 Subject: [PATCH 06/12] Polish the non-debug docs examples Finish the Python style, typing and docstring pass across the ordinary, parallelism, Trio and integration examples. Preserve each demo's runtime behavior while tightening callable, portal, stream and nursery annotations. Use the modern `.chan` portal attr and current actor-lifecycle terminology. (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- examples/a_trynamic_first_scene.py | 30 +++++-- examples/actor_spawning_and_causality.py | 12 ++- ...ctor_spawning_and_causality_with_daemon.py | 18 ++-- examples/asynchronous_generators.py | 27 ++++-- examples/full_fledged_streaming_service.py | 62 +++++++++---- examples/infected_asyncio_echo_server.py | 35 ++++++-- .../integration/mpi4py/inherit_parent_main.py | 28 +++--- examples/multiple_streams_one_portal.py | 42 ++++++--- examples/nested_actor_tree.py | 18 +++- .../parallelism/concurrent_actors_primes.py | 90 ++++++++++++++----- examples/parallelism/single_func.py | 16 ++-- examples/quick_cluster.py | 11 ++- examples/remote_error_propagation.py | 22 +++-- examples/rpc_bidir_streaming.py | 17 +++- examples/service_daemon_discovery.py | 10 ++- examples/service_discovery.py | 20 +++-- examples/streaming_broadcast_fanout.py | 13 ++- examples/trio/lockacquire_not_unmasked.py | 27 ++++-- examples/trio/send_chan_aclose_masks_beg.py | 42 ++++++--- examples/typed_payloads.py | 10 +++ examples/uds_transport_actor_tree.py | 11 ++- 21 files changed, 418 insertions(+), 143 deletions(-) diff --git a/examples/a_trynamic_first_scene.py b/examples/a_trynamic_first_scene.py index 0606e5f0..ba6d12b7 100644 --- a/examples/a_trynamic_first_scene.py +++ b/examples/a_trynamic_first_scene.py @@ -1,29 +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() -> str: + ''' + Return a greeting naming the current actor. + + ''' return the_line.format(tractor.current_actor().name) 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() -> None: - """Main tractor entry point, the "master" process (for now + ''' + Main tractor entry point, the "master" process (for now acts as the "director"). - """ + + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: - print("Alright... Action!") + print('Alright... Action!') # both actors wait on (then dial!) the *other*, so each # must outlive both hellos: spawn as daemons, run the @@ -40,6 +51,10 @@ async def main() -> None: name: str, other_actor: str, ) -> None: + ''' + Print a greeting fetched through a named actor. + + ''' print( # RPC through an existing actor's `Portal`. await portals[name].run( @@ -48,13 +63,14 @@ async def main() -> None: ) ) + tn: trio.Nursery async with trio.open_nursery() as tn: tn.start_soon(run_and_print, 'donny', 'gretchen') tn.start_soon(run_and_print, 'gretchen', 'donny') await an.cancel() - print("CUTTTT CUUTT CUT!!! Donny!! You're supposed to say...") + print('CUTTTT CUUTT CUT!!! Donny!! You\'re supposed to say...') if __name__ == '__main__': diff --git a/examples/actor_spawning_and_causality.py b/examples/actor_spawning_and_causality.py index e5a5790f..8b3dd0ac 100644 --- a/examples/actor_spawning_and_causality.py +++ b/examples/actor_spawning_and_causality.py @@ -3,13 +3,19 @@ import tractor 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() -> None: - """The main ``tractor`` routine. - """ + ''' + The main ``tractor`` routine. + + ''' # spawn a subactor, run ``cellar_door()`` as its lone task, # block until its result arrives and the subactor is reaped. print( diff --git a/examples/actor_spawning_and_causality_with_daemon.py b/examples/actor_spawning_and_causality_with_daemon.py index c75ab57f..166f08b0 100644 --- a/examples/actor_spawning_and_causality_with_daemon.py +++ b/examples/actor_spawning_and_causality_with_daemon.py @@ -3,15 +3,20 @@ import tractor async def movie_theatre_question() -> str: - """A question asked in a dark theatre, in a tangent + ''' + A question asked in a dark theatre, in a tangent (errr, I mean different) process. - """ + + ''' return 'have you ever seen a portal?' async def main() -> None: - """The main ``tractor`` routine. - """ + ''' + The main ``tractor`` routine. + + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( @@ -24,9 +29,8 @@ async def main() -> None: # 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() diff --git a/examples/asynchronous_generators.py b/examples/asynchronous_generators.py index 0202c98b..037230ff 100644 --- a/examples/asynchronous_generators.py +++ b/examples/asynchronous_generators.py @@ -1,20 +1,31 @@ -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() -> None: + ''' + Print messages streamed from a subactor. + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( @@ -24,10 +35,12 @@ async def main() -> None: # this async for loop streams values from the above # async generator running in a separate process + stream: tractor.MsgStream async with portal.open_stream_from(stream_forever) as stream: count: int = 0 - async for letter in stream: - print(letter) + message: str + async for message in stream: + print(message) count += 1 if count > 50: diff --git a/examples/full_fledged_streaming_service.py b/examples/full_fledged_streaming_service.py index de4d5286..87c777eb 100644 --- a/examples/full_fledged_streaming_service.py +++ b/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: int): +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: int): +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,6 +32,7 @@ async def aggregate(seed: int): an: ActorNursery async with tractor.open_nursery() as an: portals: list[Portal] = [] + i: int for i in range(1, 3): # fork/spawn call @@ -35,20 +43,35 @@ async def aggregate(seed: int): 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: 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: @@ -62,8 +85,9 @@ async def aggregate(seed: int): # 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) @@ -72,11 +96,11 @@ async def aggregate(seed: int): assert value in unique_vals - print("FINISHED ITERATING in aggregator") + print('FINISHED ITERATING in aggregator') await an.cancel() - print("WAITING on `ActorNursery` to finish") - print("AGGREGATOR COMPLETE!") + print('WAITING on `ActorNursery` to finish') + print('AGGREGATOR COMPLETE!') async def main() -> list[int]: @@ -95,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', @@ -109,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) diff --git a/examples/infected_asyncio_echo_server.py b/examples/infected_asyncio_echo_server.py index e111459d..b570f522 100644 --- a/examples/infected_asyncio_echo_server.py +++ b/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():`` @@ -29,8 +32,14 @@ async def aio_echo_server( 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() -> None: + ''' + Run the infected asyncio echo-server example. + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: - p: tractor.Portal = 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() -> None: break print(f'mean round trip rate (Hz): {1/mean(delays)}') - await p.cancel_actor() + await portal.cancel_actor() if __name__ == '__main__': diff --git a/examples/integration/mpi4py/inherit_parent_main.py b/examples/integration/mpi4py/inherit_parent_main.py index 741d8a6b..39d023aa 100644 --- a/examples/integration/mpi4py/inherit_parent_main.py +++ b/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,8 +33,13 @@ 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: @@ -43,9 +51,9 @@ async def main() -> None: inherit_parent_main=False, ) result: str = await portal.run(child_fn) - print(f"[parent] got: {result}", flush=True) + print(f'[parent] got: {result}', flush=True) await portal.cancel_actor() -if __name__ == "__main__": +if __name__ == '__main__': trio.run(main) diff --git a/examples/multiple_streams_one_portal.py b/examples/multiple_streams_one_portal.py index 45272b6c..81a8a3bf 100644 --- a/examples/multiple_streams_one_portal.py +++ b/examples/multiple_streams_one_portal.py @@ -1,3 +1,5 @@ +from typing import AsyncIterator + import trio import tractor @@ -5,20 +7,30 @@ import tractor log = tractor.log.get_logger('multiportal') -async def stream_data(seed: int = 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: tractor.Portal, - consumed: list, + 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) @@ -27,24 +39,32 @@ async def stream_from_portal( async def main() -> None: + ''' + Consume two concurrent streams through one portal. + ''' an: tractor.ActorNursery async with tractor.open_nursery(loglevel='info') as an: - p: tractor.Portal = await an.start_actor( + portal: tractor.Portal = await an.start_actor( 'stream_boi', enable_modules=[__name__], ) - consumed: list = [] + 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() diff --git a/examples/nested_actor_tree.py b/examples/nested_actor_tree.py index cb45b22c..279e3a4d 100644 --- a/examples/nested_actor_tree.py +++ b/examples/nested_actor_tree.py @@ -53,15 +53,21 @@ 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, @@ -85,12 +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: 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], diff --git a/examples/parallelism/concurrent_actors_primes.py b/examples/parallelism/concurrent_actors_primes.py index e38adeee..9cd7b8e6 100644 --- a/examples/parallelism/concurrent_actors_primes.py +++ b/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, @@ -32,6 +42,10 @@ PRIMES = [ 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: int) -> bool: @acm -async def worker_pool(workers: int = 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: 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,22 +95,36 @@ async def worker_pool(workers: int = 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 + ''' + # define an async (local) task to collect results from + # workers async def send_result( - func: Callable, + func: Callable[[int], Awaitable[bool]], value: int, portal: tractor.Portal, - ): - await snd_chan.send((value, await portal.run(func, n=value))) + ) -> 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, @@ -107,20 +144,29 @@ async def worker_pool(workers: int = 4): 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') diff --git a/examples/parallelism/single_func.py b/examples/parallelism/single_func.py index 9513505d..71458aa1 100644 --- a/examples/parallelism/single_func.py +++ b/examples/parallelism/single_func.py @@ -1,11 +1,11 @@ -""" +''' 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 @@ -13,18 +13,24 @@ import trio 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): - await trio.sleep(1/50000/50) + await trio.sleep(1 / 50000 / 50) return pid 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 @@ -34,7 +40,7 @@ async def main() -> None: # block on and collect its PID as the caller-side result pid: int = await tractor.to_actor.run(burn_cpu) - print(f"Collected subproc {pid}") + print(f'Collected subproc {pid}') if __name__ == '__main__': diff --git a/examples/quick_cluster.py b/examples/quick_cluster.py index 489912c1..be3597aa 100644 --- a/examples/quick_cluster.py +++ b/examples/quick_cluster.py @@ -1,10 +1,13 @@ - 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() @@ -28,7 +31,9 @@ async def main() -> None: 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, diff --git a/examples/remote_error_propagation.py b/examples/remote_error_propagation.py index 405c626b..fe399fda 100644 --- a/examples/remote_error_propagation.py +++ b/examples/remote_error_propagation.py @@ -3,17 +3,29 @@ import tractor async def assert_err() -> None: + ''' + Raise an assertion error in the current actor. + + ''' assert 0 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: list[tractor.Portal] = [] + i: int for i in range(3): - real_actors.append(await an.start_actor( - f'actor_{i}', - enable_modules=[__name__], - )) + 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!') diff --git a/examples/rpc_bidir_streaming.py b/examples/rpc_bidir_streaming.py index 68c4f541..e71007bf 100644 --- a/examples/rpc_bidir_streaming.py +++ b/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,7 +33,11 @@ async def simple_rpc( async def main() -> None: + ''' + Exercise bidirectional streaming with a remote actor. + ''' + an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( @@ -39,6 +46,9 @@ async def main() -> None: ) # 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') diff --git a/examples/service_daemon_discovery.py b/examples/service_daemon_discovery.py index 0abd4987..5c5c6b03 100644 --- a/examples/service_daemon_discovery.py +++ b/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,6 +51,10 @@ async def client_task() -> None: async def main() -> None: + ''' + Run a discoverable quote service and its client. + + ''' an: tractor.ActorNursery async with tractor.open_nursery() as an: portal: tractor.Portal = await an.start_actor( diff --git a/examples/service_discovery.py b/examples/service_discovery.py index ef678951..3d30e800 100644 --- a/examples/service_discovery.py +++ b/examples/service_discovery.py @@ -1,22 +1,28 @@ import trio import tractor -tractor.log.get_console_log("INFO") +tractor.log.get_console_log('INFO') 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) - portal: tractor.Portal - 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}' + ) - sockaddr: tractor.Portal - 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: + print(f'my_service is found at {actor_portal}') await an.cancel() diff --git a/examples/streaming_broadcast_fanout.py b/examples/streaming_broadcast_fanout.py index 15708795..c0ae45f1 100644 --- a/examples/streaming_broadcast_fanout.py +++ b/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,12 +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: 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, @@ -72,6 +82,7 @@ async def main() -> None: 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, diff --git a/examples/trio/lockacquire_not_unmasked.py b/examples/trio/lockacquire_not_unmasked.py index 19077706..58fc681d 100644 --- a/examples/trio/lockacquire_not_unmasked.py +++ b/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.TaskStatus = 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, @@ -47,6 +59,11 @@ async def main( 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( diff --git a/examples/trio/send_chan_aclose_masks_beg.py b/examples/trio/send_chan_aclose_masks_beg.py index 971d8678..669c3dea 100644 --- a/examples/trio/send_chan_aclose_masks_beg.py +++ b/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()` @@ -135,18 +140,25 @@ 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: diff --git a/examples/typed_payloads.py b/examples/typed_payloads.py index 58acada6..e70c075f 100644 --- a/examples/typed_payloads.py +++ b/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,12 +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: 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, @@ -73,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() diff --git a/examples/uds_transport_actor_tree.py b/examples/uds_transport_actor_tree.py index 6555329d..dfcfd746 100644 --- a/examples/uds_transport_actor_tree.py +++ b/examples/uds_transport_actor_tree.py @@ -21,13 +21,17 @@ 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, 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'], @@ -52,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() From 97aff520509834696b61506e42115e02141e815f Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 19:01:34 -0400 Subject: [PATCH 07/12] Polish the debugger examples Finish the Python style, typing and docstring pass across the debugger examples without changing their intentional breakpoints, failures, cancellation races or timeout reproducers. Restore full child command lines in the documented process trees and keep the examples within the 69-column source limit. (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../fast_error_in_root_after_spawn.py | 16 ++++- examples/debugging/multi_daemon_subactors.py | 27 ++++++-- ...ed_subactors_error_up_through_nurseries.py | 69 ++++++++++++------- .../debugging/multi_subactor_root_errors.py | 25 ++++--- examples/debugging/multi_subactors.py | 36 +++++++--- examples/debugging/per_actor_debug.py | 15 +++- examples/debugging/pm_in_subactor.py | 6 +- examples/debugging/root_actor_breakpoint.py | 3 + examples/debugging/root_actor_error.py | 4 ++ ...root_cancelled_but_child_is_in_tty_lock.py | 32 ++++++--- .../debugging/root_self_cancelled_w_error.py | 17 +++-- .../root_timeout_while_child_crashed.py | 15 ++-- examples/debugging/shielded_pause.py | 23 +++++-- examples/debugging/subactor_bp_in_ctx.py | 18 ++++- examples/debugging/subactor_breakpoint.py | 5 +- examples/debugging/subactor_error.py | 10 ++- 16 files changed, 233 insertions(+), 88 deletions(-) diff --git a/examples/debugging/fast_error_in_root_after_spawn.py b/examples/debugging/fast_error_in_root_after_spawn.py index a3953d36..fdafddef 100644 --- a/examples/debugging/fast_error_in_root_after_spawn.py +++ b/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__], ) @@ -36,7 +43,10 @@ async def open_ctx( async def main() -> None: + ''' + Fail the root while a subactor context is still starting. + ''' async with tractor.open_nursery( debug_mode=True, loglevel='runtime', diff --git a/examples/debugging/multi_daemon_subactors.py b/examples/debugging/multi_daemon_subactors.py index 95822f93..0c77ffed 100644 --- a/examples/debugging/multi_daemon_subactors.py +++ b/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,8 +20,11 @@ async def breakpoint_forever(): raise -async def name_error(): - "Raise a ``NameError``" +async def name_error() -> None: + ''' + Raise a ``NameError``. + + ''' getattr(doggypants) # noqa @@ -28,8 +36,14 @@ async def main() -> None: 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() -> None: except tractor.RemoteActorError as rae: assert rae.boxed_type is NameError + i: str async for i in stream: # a second time try the failing subactor and this tie diff --git a/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py b/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py index 2895f0e6..6357b96f 100644 --- a/examples/debugging/multi_nested_subactors_error_up_through_nurseries.py +++ b/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( @@ -64,24 +75,34 @@ async def spawn_until(depth=0): # TODO: notes on the new boxed-relayed errors through proxy actors async def main() -> None: - """The main ``tractor`` routine. + ''' + 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, diff --git a/examples/debugging/multi_subactor_root_errors.py b/examples/debugging/multi_subactor_root_errors.py index b934c515..a9d0debb 100644 --- a/examples/debugging/multi_subactor_root_errors.py +++ b/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, @@ -24,7 +29,8 @@ async def spawn_error(): async def main() -> None: - """The main ``tractor`` routine. + ''' + The main ``tractor`` routine. The process tree should look as approximately as follows: @@ -37,7 +43,8 @@ async def main() -> None: - nested name_error sub-sub-actor - root actor should then fail on assert - program termination - """ + + ''' async with ( tractor.open_nursery( debug_mode=True, @@ -46,11 +53,11 @@ async def main() -> None: 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__], ) diff --git a/examples/debugging/multi_subactors.py b/examples/debugging/multi_subactors.py index f2ea18c9..efa8bfd1 100644 --- a/examples/debugging/multi_subactors.py +++ b/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, @@ -26,7 +36,8 @@ async def spawn_error(): async def main() -> None: - """The main ``tractor`` routine. + ''' + The main ``tractor`` routine. The process tree should look as approximately as follows: @@ -35,7 +46,8 @@ async def main() -> None: |-python -m tractor._child --uid ('bp_forever', '1f787a7e ...) `-python -m tractor._child --uid ('spawn_error', '52ee14a5 ...) `-python -m tractor._child --uid ('name_error', '3391222c ...) - """ + + ''' errors: list[BaseException] = [] async with tractor.open_nursery( @@ -43,12 +55,14 @@ async def main() -> None: # 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). ''' diff --git a/examples/debugging/per_actor_debug.py b/examples/debugging/per_actor_debug.py index c5abe450..8345a35f 100644 --- a/examples/debugging/per_actor_debug.py +++ b/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() -> 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, diff --git a/examples/debugging/pm_in_subactor.py b/examples/debugging/pm_in_subactor.py index a9728a6b..cc60347d 100644 --- a/examples/debugging/pm_in_subactor.py +++ b/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. @@ -49,7 +49,9 @@ async def main() -> None: 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__': diff --git a/examples/debugging/root_actor_breakpoint.py b/examples/debugging/root_actor_breakpoint.py index 347123ef..35a029ec 100644 --- a/examples/debugging/root_actor_breakpoint.py +++ b/examples/debugging/root_actor_breakpoint.py @@ -3,7 +3,10 @@ import tractor async def main() -> None: + ''' + Pause in the root actor to exercise its debugger REPL. + ''' async with tractor.open_root_actor( debug_mode=True, ): diff --git a/examples/debugging/root_actor_error.py b/examples/debugging/root_actor_error.py index 49359f16..1fae47b3 100644 --- a/examples/debugging/root_actor_error.py +++ b/examples/debugging/root_actor_error.py @@ -3,6 +3,10 @@ import tractor async def main() -> None: + ''' + Raise an assertion error from the debug-enabled root actor. + + ''' async with tractor.open_root_actor( debug_mode=True, ): diff --git a/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py b/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py index ca15530b..7fafa1bb 100644 --- a/examples/debugging/root_cancelled_but_child_is_in_tty_lock.py +++ b/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) @@ -33,12 +40,17 @@ async def main() -> None: 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 ( diff --git a/examples/debugging/root_self_cancelled_w_error.py b/examples/debugging/root_self_cancelled_w_error.py index e5ebbacc..fb4b2ecb 100644 --- a/examples/debugging/root_self_cancelled_w_error.py +++ b/examples/debugging/root_self_cancelled_w_error.py @@ -3,6 +3,10 @@ import tractor 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() -> None: 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 diff --git a/examples/debugging/root_timeout_while_child_crashed.py b/examples/debugging/root_timeout_while_child_crashed.py index 11533a8e..8ca1a813 100644 --- a/examples/debugging/root_timeout_while_child_crashed.py +++ b/examples/debugging/root_timeout_while_child_crashed.py @@ -2,8 +2,11 @@ import trio import tractor -async def key_error(): - "Raise a ``NameError``" +async def key_error() -> None: + ''' + Raise a ``KeyError``. + + ''' return {}['doggy'] @@ -21,7 +24,7 @@ async def main() -> None: 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() -> None: # 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() diff --git a/examples/debugging/shielded_pause.py b/examples/debugging/shielded_pause.py index cfa8f4f8..a145a3bb 100644 --- a/examples/debugging/shielded_pause.py +++ b/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. @@ -72,6 +83,10 @@ async def cancelled_before_pause( async def main() -> None: + ''' + Exercise shielded debugger entry in subactor and root tasks. + + ''' async with tractor.open_nursery( debug_mode=True, ) as an: diff --git a/examples/debugging/subactor_bp_in_ctx.py b/examples/debugging/subactor_bp_in_ctx.py index eafeb0b7..3b3074c2 100644 --- a/examples/debugging/subactor_bp_in_ctx.py +++ b/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) @@ -35,14 +44,17 @@ async def just_bp( 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() -> None: enable_transports=[tpt], loglevel='devx', ) as an: - p = await an.start_actor( + p: tractor.Portal = await an.start_actor( 'bp_boi', enable_modules=[__name__], ) diff --git a/examples/debugging/subactor_breakpoint.py b/examples/debugging/subactor_breakpoint.py index 0a047a21..7b844f10 100644 --- a/examples/debugging/subactor_breakpoint.py +++ b/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. @@ -13,7 +13,10 @@ async def breakpoint_forever(): async def main() -> None: + ''' + Run a subactor that repeatedly pauses in the debugger. + ''' async with tractor.open_nursery( debug_mode=True, loglevel='cancel', diff --git a/examples/debugging/subactor_error.py b/examples/debugging/subactor_error.py index fd280cb9..8ec65781 100644 --- a/examples/debugging/subactor_error.py +++ b/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() -> None: + ''' + Surface a subactor `NameError` at the waiting root task. + + ''' async with tractor.open_nursery( debug_mode=True, ) as an: From 38883dca0317330f6d7effe7a5f85a8034aaab02 Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 20:05:28 -0400 Subject: [PATCH 08/12] Harden the dedicated registrar example Run the registrar in its own process and prove that a sibling client discovers the service through registry lookup instead of an existing peer channel. Retry ephemeral bind collisions, publish readiness atomically and validate bounded cross-platform shutdown. Document actual duplicate name and multi-registrar ordering semantics alongside the example. Move the demo under the discovery examples and wrap process ownership in an `@acm`. Record the future public subsystem, Piker service and pytest isolation follow-ups. (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- docs/guide/discovery.rst | 85 +++-- examples/dedicated_registrar.py | 122 ------- examples/discovery/dedicated_registrar.py | 394 ++++++++++++++++++++++ 3 files changed, 450 insertions(+), 151 deletions(-) delete mode 100644 examples/dedicated_registrar.py create mode 100644 examples/discovery/dedicated_registrar.py diff --git a/docs/guide/discovery.rst b/docs/guide/discovery.rst index 4ce2d6ad..a14967e0 100644 --- a/docs/guide/discovery.rst +++ b/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? -------------------------------- @@ -69,18 +69,33 @@ 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. Boot -a bare ``tractor.run_daemon([], registry_addrs=[...])`` (a root -actor that does nothing but hold the registry), point your app -tree at the same ``registry_addrs``, and every actor discovers -through that *external* registrar instead of a tree-local one: +**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/dedicated_registrar.py - :caption: examples/dedicated_registrar.py +.. literalinclude:: ../../examples/discovery/dedicated_registrar.py + :caption: examples/discovery/dedicated_registrar.py :language: python -This is the "registrar as a subsystem, not the root actor" shape. -Two caveats today (both tracked as #472 follow-ups): +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 @@ -114,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()`` ******************** @@ -153,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: @@ -223,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 ------------------------------- @@ -248,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): @@ -276,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 diff --git a/examples/dedicated_registrar.py b/examples/dedicated_registrar.py deleted file mode 100644 index d80af49b..00000000 --- a/examples/dedicated_registrar.py +++ /dev/null @@ -1,122 +0,0 @@ -''' -Run a *dedicated* registrar as its own standalone process — decoupled -from your app's root actor — and discover a service *through* it. - -Normally the registrar **is** the root actor of your tree. Here we -instead boot a separate `tractor.run_daemon([], registry_addrs=[...])` -process whose *sole* job is to be the registry, then point our app -tree at it via `registry_addrs`. Because a registrar is already -reachable at that addr, our app's root actor does NOT become one — it -registers with (and discovers through) the external daemon. That's the -"registrar as a subsystem, not the root actor" pattern. - -NB: `enable_transports` is single-proto per-runtime today (see -`tractor._root`), so this demos one transport; a genuinely -multi-backend registrar (and spawning one as a *sub*actor of a shared -tree) are future runtime work — see the #472 follow-ups. - -''' -from contextlib import suppress -import signal -import socket -import subprocess -import sys -import time - -import trio -import tractor - - -# the fixed addr the dedicated registrar binds and everyone points at. -REG_ADDR: tuple[str, int] = ('127.0.0.1', 1717) - - -def _wait_registrar_ready( - addr: tuple[str, int], - proc: subprocess.Popen, - deadline: float = 10.0, -) -> None: - ''' - Active-poll the registrar's bind addr until it accepts a - connection (proving it's booted + listening), bailing early if - the daemon proc dies during startup. - - ''' - end: float = time.monotonic() + deadline - while time.monotonic() < end: - if proc.poll() is not None: - raise RuntimeError( - f'registrar died on startup (rc={proc.returncode})' - ) - with suppress(OSError): - with socket.create_connection(addr, timeout=0.1): - return - time.sleep(0.05) - raise TimeoutError(f'registrar never came up @ {addr}') - - -async def greet() -> str: - '''A trivial service task any peer can RPC by name.''' - return f'hello from {tractor.current_actor().name}!' - - -async def app() -> None: - ''' - Point our app tree at the EXTERNAL registrar (not its own root) - via `registry_addrs`, register a named service, then discover + - RPC it purely by name. - - ''' - an: tractor.ActorNursery - async with tractor.open_nursery( - registry_addrs=[REG_ADDR], - enable_transports=['tcp'], - ) as an: - # this subactor registers with the DEDICATED registrar @ - # REG_ADDR (our root is a plain peer, not the registry). - await an.start_actor( - 'greeter', - enable_modules=[__name__], - ) - # discover it *through the external registrar*, by name only. - portal: tractor.Portal - async with tractor.wait_for_actor('greeter') as portal: - print(f'found `greeter` via dedicated registrar @ {REG_ADDR}') - print(await portal.run(greet)) - await an.cancel() - - -def main() -> None: - # boot the dedicated registrar as its own process/tree: an empty - # `enable_modules` `run_daemon()` is just a root actor that does - # nothing but hold + serve the registry. - code: str = ( - 'import tractor; ' - f'tractor.run_daemon([], registry_addrs={[REG_ADDR]!r}, ' - "enable_transports=['tcp'], loglevel='error')" - ) - registrar: subprocess.Popen = subprocess.Popen( - [sys.executable, '-c', code], - # the registry is a quiet background service; hush its logs + - # expected SIGINT-teardown traceback so the demo output stays - # focused on the discovery flow. - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) - try: - _wait_registrar_ready(REG_ADDR, registrar) - print( - f'dedicated registrar up @ {REG_ADDR} ' - f'(pid {registrar.pid})' - ) - trio.run(app) - finally: - # graceful SIGINT teardown of the standalone registrar. - registrar.send_signal(signal.SIGINT) - with suppress(subprocess.TimeoutExpired): - registrar.wait(timeout=10) - print('dedicated registrar shut down') - - -if __name__ == '__main__': - main() diff --git a/examples/discovery/dedicated_registrar.py b/examples/discovery/dedicated_registrar.py new file mode 100644 index 00000000..f4386f13 --- /dev/null +++ b/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) From a40fb2ebde467fabbe019788ac501971c76ba9dd Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 20:07:06 -0400 Subject: [PATCH 09/12] Correct typed-msg validation docs Document `Started` as the eager sender-side payload check and `Yield` plus `Return` as receiver-side decoding boundaries without promising a symmetric error relay. Separate the working task-scoped codec encoder from the private per-dialog decoder and the incomplete `@context` hook params. Link the planned typed `Start` contract and sender-side argument validation follow-up in #514. (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- docs/guide/msging.rst | 72 ++++++++++++++++++++++++------------------- 1 file changed, 41 insertions(+), 31 deletions(-) diff --git a/docs/guide/msging.rst b/docs/guide/msging.rst index eb5f8b8d..d0760eb9 100644 --- a/docs/guide/msging.rst +++ b/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 @@ -253,13 +260,15 @@ escape hatch. Both are exercised end-to-end in ``tests/msg/test_pldrx_limiting.py`` and ``tests/msg/test_ext_types_msgspec.py``. -On the codec-hook side, the ``enc_hook``/``dec_hook`` pair is -today only reachable via ``tractor.msg._ops``; a public *factory* -API for them is drafted in `#376`_ (from +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 `_, on the `auto_codecs `_ -branch) — the likely long-term home for custom-type -(de)serialization. +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. @@ -282,3 +291,4 @@ Where to next? .. _#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 From a95ca7577f21d511348672fa323ea874863d2213 Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 20:10:13 -0400 Subject: [PATCH 10/12] Record PR #512 prompt provenance Capture the human direction, generated changes, review findings and validation results for the docs-example landing pass. Point generated-code references at the complete commit range from the pre-remediation branch head. Prompt-IO: ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../20260828T200822Z_0be872ff_prompt_io.md | 62 +++++++++++++ ...20260828T200822Z_0be872ff_prompt_io.raw.md | 90 +++++++++++++++++++ 2 files changed, 152 insertions(+) create mode 100644 ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.raw.md diff --git a/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.md b/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.md new file mode 100644 index 00000000..6304e7e8 --- /dev/null +++ b/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. diff --git a/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.raw.md b/ai/prompt-io/opencode/20260828T200822Z_0be872ff_prompt_io.raw.md new file mode 100644 index 00000000..554459f5 --- /dev/null +++ b/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. From d799bd6f684a6f5c909ec82f82d6771648549b8f Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 20:24:25 -0400 Subject: [PATCH 11/12] Show the discovered actor address Print the discovered service's remote channel address instead of the default `Portal` object repr. Review: https://github.com/goodboy/tractor/pull/512#discussion_r3883250916 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- examples/service_discovery.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/examples/service_discovery.py b/examples/service_discovery.py index 3d30e800..5787359a 100644 --- a/examples/service_discovery.py +++ b/examples/service_discovery.py @@ -22,7 +22,8 @@ async def main(service_name: str) -> None: async with tractor.wait_for_actor( service_name, ) as actor_portal: - print(f'my_service is found at {actor_portal}') + service_addr = actor_portal.chan.raddr + print(f'my_service is found at {service_addr}') await an.cancel() From 77964800624234db7ed50962d31dbefe26b63ab8 Mon Sep 17 00:00:00 2001 From: goodboy Date: Sat, 29 Aug 2026 21:46:38 -0400 Subject: [PATCH 12/12] Widen `accept_addr` in the UDS example Match `Actor.accept_addr`'s declared `tuple[str, int|str]` contract instead of narrowing its second item to the UDS-specific string path. Review: https://github.com/goodboy/tractor/pull/512#discussion_r3888153533 (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- examples/uds_transport_actor_tree.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/uds_transport_actor_tree.py b/examples/uds_transport_actor_tree.py index dfcfd746..68b780ac 100644 --- a/examples/uds_transport_actor_tree.py +++ b/examples/uds_transport_actor_tree.py @@ -22,7 +22,7 @@ async def report_addr() -> str: ''' actor: tractor.Actor = tractor.current_actor() - addr: tuple[str, str] = actor.accept_addr + addr: tuple[str, int|str] = actor.accept_addr pid: int = os.getpid() return f'{actor.name}@{addr} pid={pid}'