Hoist ria-reaping out of the spawn backends

Step A of the `._ria_nursery` removal (issue #477 follow-up, see
`ai/conc-anal/ria_nursery_removal_plan.md`): `.run_in_actor()`
children now spawn via the default daemon nursery and their
result-reaping moves up into the `ActorNursery` machinery,

- new `_supervise._reap_ria_portals()`: one
  `_spawn.cancel_on_completion()` task per ria child, run AFTER
  `._join_procs` is set — replacing the per-child reaper task the
  backends formerly spawned (keyed off
  `._cancel_after_result_on_exit` membership) which required
  routing such children into `._ria_nursery`.
- happy path: reap awaited right after `._join_procs.set()`,
  preserving "collect ria results before daemon join" sequencing.
- error path: snapshot ria `(portal, subactor)` pairs (backend
  `finally`s pop `._children` as procs reap), `await an.cancel()`,
  THEN a 0.5s-bounded reap over the snapshot; anything collectable
  is already queued in the local ctx and a parked reaper
  self-cleans (`trio.Cancelled` results are never stashed). NB: a
  concurrent reap+cancel variant deadlocked `test_multierror` and
  a 3s bound blew `test_cancel_while_childs_child_in_sync_sleep`'s
  deadline — deats in the plan doc's probe history.
- `spawn/_trio.py` + `spawn/_mp.py`: drop the membership branch,
  per-child reaper nursery + now-unused `cancel_on_completion`
  imports; the join phase is a bare `soft_kill()`.

`._ria_nursery` is now vestigial (zero spawn users): step B
deletes it + `start_actor()`'s `nursery=` escape hatch and merges
the supervisor's two error handlers.

Prompt-IO: ai/prompt-io/claude/20260702T165806Z_a34aaf98_prompt_io.md

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

ai/conc-anal/ria_nursery_removal_plan.md

@@ -0,0 +1,187 @@
+# `_ria_nursery` removal plan (issue #477 follow-up)
+
+Goal: drop the secondary "run-in-actor" spawn nursery (and
+friends) from `ActorNursery`/spawn internals, now that
+`tractor.to_actor.run()` delivers one-shot semantics purely on
+the daemon-spawn + portal primitives.
+
+## Verified machinery map (2026-07-02, wkt @ a34aaf98)
+
+The entire mechanism is 4 files:
+
+- `runtime/_supervise.py`
+  - `ActorNursery.__init__(.., ria_nursery, ..)` stores
+    `._ria_nursery` (:202, :238); sole read is
+    `run_in_actor()` passing `nursery=self._ria_nursery`
+    (:442) into `start_actor()`'s `nursery:
+    trio.Nursery|None` escape-hatch param (:305, :367).
+  - `._cancel_after_result_on_exit: set` (:244) marks ria
+    portals (:457).
+  - `_open_and_supervise_one_cancels_all_nursery()` nests
+    `da_nursery` (:609) around `ria_nursery` (:622); the
+    `finally:` at the ria->da boundary (:747-766) raises
+    collected `errors` (single exc or BEG).
+- `runtime/_portal.py`
+  - `._expect_result_ctx` (:112) set by `_submit_for_result()`
+    (:142, sole caller `run_in_actor()`); consumed by
+    `wait_for_result()` (:167) + deprecated `result()` (:220).
+    The `None` branch (:184-196) returns the `NoResult`
+    sentinel (`_exceptions.py:1164`).
+- `spawn/_spawn.py`
+  - `exhaust_portal()` (:129): awaits
+    `portal.wait_for_result()`, CATCHES+RETURNS any exc
+    (never raises).
+  - `cancel_on_completion()` (:177): `exhaust_portal()` ->
+    on exc-result stash `errors[uid] = result` (:203) ->
+    ALWAYS `portal.cancel_actor()` (:218).
+- `spawn/_trio.py` (:195-222) + `spawn/_mp.py` (:187-213),
+  identical shape: after shielded
+  `await an._join_procs.wait()`, open a per-child local
+  nursery; IFF `portal in an._cancel_after_result_on_exit`
+  start `cancel_on_completion` alongside `soft_kill()`; when
+  `soft_kill` returns first, `nursery.cancel_scope.cancel()`
+  reaps the result-waiter.
+
+## The load-bearing semantic (already-deferred errors)
+
+Remote ria-child errors NEVER raise into `ria_nursery`:
+
+1. reaper tasks only START after `_join_procs.set()` (block
+   exit or the inner error handler),
+2. `exhaust_portal` swallows the exc into a return value,
+3. `cancel_on_completion` stashes it in `errors` + cancels
+   that child,
+4. the ria->da `finally:` re-raises collected `errors` (and
+   `an.cancel()`s any daemon stragglers).
+
+So mid-block there is NO error propagation from ria children
+(unless user code explicitly `await portal.wait_for_result()`s)
+— the two-nursery nesting only sequences "reap ria results
+BEFORE blocking on daemon join". A single-nursery impl only
+needs to preserve that sequencing, not any ASAP-cancel
+behavior.
+
+## Target design
+
+### step A: single-nursery `run_in_actor()` (mechanical)
+
+- `run_in_actor()` spawns via the DEFAULT (`_da_nursery`)
+  path — drop `nursery=self._ria_nursery`.
+- rename `._cancel_after_result_on_exit` ->
+  `._ria_portals: dict[portal, Actor]` (need the subactor ref
+  for `cancel_on_completion`).
+- move reaper start-up OUT of the backends into
+  `_open_and_supervise...`: immediately after EACH
+  `an._join_procs.set()` call-site (happy path :642, inner
+  error handler :661), start one
+  `cancel_on_completion(portal, subactor, errors)` task per
+  ria portal into `da_nursery`, then (happy path only)
+  `await` their completion BEFORE falling out of the
+  `try:`/`finally:` that raises `errors` — e.g. gather in a
+  dedicated inner `trio.open_nursery()` block replacing
+  today's `ria_nursery` join point.
+- delete the membership branch + local reaper nursery from
+  `_trio.py`/`_mp.py` (keep the `soft_kill()` call; the
+  per-child local nursery collapses to just `soft_kill`).
+- `_trio.py:310` `_children.pop()` etc. unchanged.
+
+### step B: delete the plumbing
+
+- `_open_and_supervise...`: drop the inner
+  `ria_nursery` + merge its `except BaseException` classify
+  logic into ONE handler on the (now single) nursery scope;
+  `ActorNursery.__init__` loses the `ria_nursery` param.
+- `start_actor()` loses the `nursery:` escape-hatch param
+  (the :302-304 TODO).
+- backends: no more `_cancel_after_result_on_exit` refs.
+
+### step C: (separate PRs) deprecate + migrate + excise
+
+- migrate in-repo `.run_in_actor()` usage to
+  `to_actor.run()`: tests 46 hits/9 files (test_cancellation
+  15, test_infected_asyncio 10, test_spawning 8, registrar 3,
+  adv_streaming 4, pubsub 2, rpc 1, runtime 1), examples 28
+  hits/13 files (debugging/* dominate), docs 20 hits/8 rst
+  files. NOTE: many sites also use deprecated
+  `Portal.result()`/`wait_for_result()` — these die with
+  `_expect_result_ctx`, so migration must land FIRST.
+- add `DeprecationWarning` to `run_in_actor()` (+
+  `_submit_for_result`/`wait_for_result`).
+- final excision: `run_in_actor()`, `_submit_for_result`,
+  `_expect_result_ctx`, `wait_for_result`/`result`,
+  `exhaust_portal`, `cancel_on_completion`, `NoResult`.
+
+## Risk register
+
+1. hard-killed ria child: today the backend-local
+   `nursery.cancel_scope.cancel()` discards a still-parked
+   reaper when the proc dies first; a da_nursery-hosted
+   reaper instead sees the transport break ->
+   `exhaust_portal` returns a `TransportClosed`-ish exc ->
+   NEW entry in `errors` that today gets discarded. Guard:
+   reap-gather block must cancel remaining reapers once all
+   ria procs are dead, or filter transport-death excs for
+   already-`cancel_called` children.
+2. error-path ordering: inner handler today sets
+   `_join_procs` THEN `an.cancel()`; reapers race the
+   cancel-RPC. Keep that ordering when moving reaper spawn.
+3. debugger interplay: `maybe_wait_for_debugger()` calls
+   (:654, :730) must stay BEFORE any reap/cancel issuance.
+4. `errors` double-entry: local body error (:646) + child's
+   relayed exc (via reaper) can both land for the same
+   scenario -> BEG shape changes vs today? (today has the
+   same dual-write sites; keep behavior identical.)
+5. mp backend parity: mirror every `_trio.py` edit in
+   `_mp.py` (identical block).
+
+## Step-A first-probe findings (2026-07-02, WIP in tree)
+
+Step A is IMPLEMENTED (uncommitted):
+`run_in_actor()` spawns via da_nursery; new
+`_supervise._reap_ria_portals()` helper; reap awaited after
+happy-path `_join_procs.set()`; error-path runs reap
+CONCURRENT with `an.cancel()` in the shielded block;
+backends stripped of the membership branch + per-child
+reaper nursery (+ dead imports).
+
+Probe history (trio backend):
+- `tests/test_to_actor.py` + `tests/test_spawning.py`:
+  20/20 PASS — incl. all `run_in_actor()` result
+  round-trips + `test_remote_error` (single erroring child,
+  body re-raise -> inner error path).
+- FIRST attempt ran the error-path reap CONCURRENT with
+  `an.cancel()` (mimicking the old backend-side race):
+  `test_cancellation.py::test_multierror` (2 erroring ria
+  children, body re-raises one) DEADLOCKED. Root cause per
+  the sequencing fix below: reap + cancel must NOT race at
+  this layer (suspected `._children` pop-during-iteration
+  and/or double-cancel RPC wedge; not fully root-caused
+  since the fix removes the race wholesale).
+- FIX (2nd attempt, current impl): error path SEQUENCES:
+  (1) snapshot ria `(portal, subactor)` pairs (backend
+  `finally`s pop `._children` as procs reap), (2)
+  `await an.cancel()`, (3) bounded reap over the snapshot.
+  Bound was first 3s -> blew the `fail_after` deadline in
+  `test_cancel_while_childs_child_in_sync_sleep` (hard-
+  killed grandchild never relays => reaper parks the full
+  bound). Tightened to 0.5s: anything collectable is
+  already queued in the local ctx (relayed BEFORE the
+  cancel); a parked reaper self-cleans (`trio.Cancelled`
+  results are never stashed).
+- RESULT: `tests/test_cancellation.py` FULLY GREEN
+  (20 passed, 1 xfailed, 77s); full-suite gate run kicked
+  off same session (see final report/next session).
+
+Remaining risk: on slow CI a relayed-but-undelivered error
+racing the 0.5s bound could drop an `errors` entry
+(BEG-shape flake); if observed, scale the bound via the
+`cpu_perf_headroom()`-style approach or peek
+`Portal._final_result_msg`/ctx queue state instead of
+time-bounding.
+
+## Verification gate
+
+- `tests/test_cancellation.py test_spawning.py test_local.py
+  test_rpc.py` on `trio` + `mp_spawn` + `mp_forkserver`
+  backends, then full suite; `tests/devx/test_debugger.py`
+  for risk 3.

ai/prompt-io/claude/20260702T154255Z_65bf9df5_prompt_io.md

@@ -0,0 +1,79 @@
+---
+model: claude-fable-5
+service: claude
+session: f6c84722-471a-4458-9a80-e453fea9029f
+timestamp: 2026-07-02T15:42:55Z
+git_ref: 65bf9df5
+scope: code
+substantive: true
+raw_file: 20260702T154255Z_65bf9df5_prompt_io.raw.md
+---
+
+## Prompt
+
+Driver prompt file `ai/prompt-io/prompts/issue_477.md`:
+
+> attempt to resolve
+> https://github.com/goodboy/tractor/issues/477
+> do it with /open-wkt.
+
+(plus a hard stop-for-human-review deadline of 12:50PM
+EST the same day)
+
+Issue #477 asks to factor `ActorNursery.run_in_actor()`
+(and possibly `Portal.run()`) out of the nursery
+internals into a new `tractor.to_actor` wrapper
+subpackage of "higher level one shot" single-remote-task
+APIs, adopting the `trio.to_thread`/`anyio.to_process`
+parlance, so that error collection/propagation moves up
+into the caller's local `trio` scope and the nursery's
+spawn machinery can eventually drop the
+`._ria_nursery` coupling.
+
+## Response summary
+
+First-cut `tractor.to_actor` subpkg delivering the
+one-shot API composed purely from the existing
+daemon-spawn + portal primitives (`start_actor()` +
+`Portal.run()` + `Portal.cancel_actor()`), leaving the
+legacy `.run_in_actor()` machinery untouched (formal
+deprecation deferred until in-repo usage migrates):
+
+- `to_actor.run(fn, **fn_kwargs) -> Any`: spawn a
+  subactor, schedule `fn` as its lone remote task, wait
+  on and return its result, ALWAYS reaping the subactor
+  (shield-safe `finally`). Remote errors raise in the
+  caller's task as boxed `RemoteActorError`s.
+- placement variants: `portal=` reuses a running actor
+  (no spawn/reap), `an=` spawns from a caller-managed
+  actor-nursery, neither opens a call-scoped private
+  `open_nursery()` (implicitly booting the runtime,
+  configurable via `runtime_kwargs`).
+- fail-fast validation before any spawn: non-streaming
+  async fn required; `portal=`/`an=` mutually
+  exclusive; `runtime_kwargs` rejected alongside any
+  placement opt.
+- `run_in_actor()` TODO/docstring now cross-reference
+  the successor API.
+
+## Files changed
+
+- `tractor/to_actor/__init__.py` — new subpkg,
+  re-exports `run`
+- `tractor/to_actor/_api.py` — `run()` +
+  `_invoke_in_subactor()` + `_validate_one_shot_fn()`
+- `tractor/__init__.py` — top-level `to_actor`
+  re-export
+- `tractor/runtime/_supervise.py` — comment/docstring
+  pointers from `run_in_actor()` to the successor
+- `tests/test_to_actor.py` — 11-test suite covering
+  all placement variants, error relay, the concurrent
+  worker-pool-ish pattern and arg validation
+- `examples/parallelism/to_actor_one_shots.py` —
+  runnable demo (auto-collected by
+  `test_docs_examples.py`)
+
+## Human edits
+
+None yet — pending human review (work paused before the
+12:50PM EST deadline per the driver prompt).

ai/prompt-io/claude/20260702T154255Z_65bf9df5_prompt_io.raw.md

@@ -0,0 +1,100 @@
+---
+model: claude-fable-5
+service: claude
+timestamp: 2026-07-02T15:42:55Z
+git_ref: 65bf9df5
+diff_cmd: git diff main..wkt/to_actor_subpkg
+---
+
+# Raw AI output (diff-ref mode)
+
+All generated code is committed on the
+`wkt/to_actor_subpkg` branch; per diff-ref mode each
+file's verbatim content is reachable via the pointers
+below rather than duplicated here.
+
+## Generated files
+
+> `git diff main..wkt/to_actor_subpkg -- tractor/to_actor/__init__.py`
+
+New subpackage init: module docstring establishing the
+`trio.to_thread`/`anyio.to_process` "run it over there"
+parlance for actors, plus the single public re-export
+`run as run` from `._api`.
+
+> `git diff main..wkt/to_actor_subpkg -- tractor/to_actor/_api.py`
+
+The one-shot invocation impl, composed entirely from the
+lower level daemon-spawn + portal primitives as
+prescribed by issue #477:
+
+- `_validate_one_shot_fn()`: the `Portal.run()`
+  non-streaming-async-fn constraint checked up-front,
+  before any subactor is spawned.
+- `_invoke_in_subactor()`: `an.start_actor()` ->
+  `Portal.run()` -> always-reap via
+  `Portal.cancel_actor()` in a `finally` (the cancel
+  req's bounded wait is internally shielded so the reap
+  also runs under caller-scope cancellation).
+- `run()`: the public API. Placement options:
+  `portal=` (reuse a running actor, no spawn/reap),
+  `an=` (spawn from a caller-managed nursery), or
+  neither (private `open_nursery()` scoped to the call,
+  implicitly booting the runtime when needed, tunable
+  via pass-through `runtime_kwargs`). Spawn opts mirror
+  `ActorNursery.start_actor()`; `**fn_kwargs` are
+  relayed to the remote task. Errors raise in the
+  caller's task as boxed `RemoteActorError`s.
+  `runtime_kwargs` alongside any placement opt is a
+  hard `ValueError`, never silently ignored.
+
+> `git diff main..wkt/to_actor_subpkg -- tractor/__init__.py`
+
+Top-level `from . import to_actor as to_actor`
+re-export.
+
+> `git diff main..wkt/to_actor_subpkg -- tractor/runtime/_supervise.py`
+
+Comment/docstring-only: the `run_in_actor()` deprecation
+TODO now points at the implemented `.to_actor.run()`
+successor (checkbox ticked) and the method docstring
+gains a NOTE steering users to the new API; remaining
+TODO items are the `DeprecationWarning` emission +
+in-repo usage migration.
+
+> `git diff main..wkt/to_actor_subpkg -- tests/test_to_actor.py`
+
+11-test suite: private-nursery one-shot, implicit
+runtime boot via `runtime_kwargs`, remote-error relay to
+the caller's task (bare + caller-managed nursery),
+caller-nursery spawn, portal reuse w/o implicit reap,
+the concurrent worker-pool-ish pattern (local `trio`
+nursery x shared `an`), and the four validation
+rejections (sync fn, async-gen fn, `portal`+`an`
+combo, `runtime_kwargs`+placement combo).
+
+> `git diff main..wkt/to_actor_subpkg -- examples/parallelism/to_actor_one_shots.py`
+
+Runnable example (auto-collected by
+`test_docs_examples.py`): the fully-implicit one-shot
+plus the concurrent worker-pool-ish prime-check pattern
+against a shared caller-managed actor-nursery.
+
+## Test runs (verbatim)
+
+```
+tests/test_to_actor.py ..........                [100%]
+============= 10 passed in 4.29s =============
+```
+
+Regression subset for touched modules
+(`test_local.py test_rpc.py test_spawning.py
+test_cancellation.py`):
+
+```
+38 passed, 1 xfailed, 24 warnings in 80.71s (0:01:20)
+```
+
+(warnings are pre-existing stdlib `os.fork()`
+DeprecationWarnings from the mp spawn backends, not
+introduced by this change)

ai/prompt-io/claude/20260702T165806Z_a34aaf98_prompt_io.md

@@ -0,0 +1,78 @@
+---
+model: claude-fable-5
+service: claude
+session: f6c84722-471a-4458-9a80-e453fea9029f
+timestamp: 2026-07-02T16:58:06Z
+git_ref: a34aaf98
+scope: code
+substantive: true
+raw_file: 20260702T165806Z_a34aaf98_prompt_io.raw.md
+---
+
+## Prompt
+
+Follow-up round in the same session as the
+`tractor.to_actor` landing (see
+`20260702T154255Z_65bf9df5_prompt_io.md`). After
+committing that work the user green-lit the deferred
+items:
+
+> go go go on this with what time you have left, in
+> particular see if you can get the _ria_nursery
+> removal going!
+
+then extended the deadline twice to iterate on the
+discovered hang:
+
+> continue on this up until a 12:58:30 deadline
+
+and finally chose "Commit step A now" from the
+next-steps prompt.
+
+## Response summary
+
+Step A of the `._ria_nursery` removal (issue #477): hoist
+`.run_in_actor()` result-reaping out of the spawn
+backends into the `ActorNursery` machinery so ria
+children spawn via the default daemon nursery,
+
+- new `_supervise._reap_ria_portals()` runs one
+  `_spawn.cancel_on_completion()` task per ria child
+  AFTER `._join_procs` is set; happy path awaits it
+  right after `._join_procs.set()`.
+- error path SEQUENCES: snapshot ria
+  `(portal, subactor)` pairs -> `await an.cancel()` ->
+  0.5s-bounded reap. Two failed intermediates informed
+  this: a concurrent reap+cancel DEADLOCKED
+  `test_multierror`; a 3s bound blew
+  `test_cancel_while_childs_child_in_sync_sleep`'s
+  `fail_after` deadline.
+- backends (`spawn/_trio.py`, `spawn/_mp.py`) lose the
+  `._cancel_after_result_on_exit` membership branch,
+  per-child reaper nursery + dead imports.
+- design/probe-history doc:
+  `ai/conc-anal/ria_nursery_removal_plan.md` (from an
+  agent-verified machinery map).
+
+Verification: `test_cancellation.py` fully green
+(20 passed, 1 xfailed) incl. the previously-hung
+`test_multierror`; `test_to_actor`+`test_spawning`
+20/20; bounded full-suite gate SIGINT'd ~30s early at
+303 passed / 0 failures (user opted to commit on that
+signal, deferring the unbounded re-run to step-B
+verification).
+
+## Files changed
+
+- `tractor/runtime/_supervise.py` — `_reap_ria_portals()`
+  + two call-sites; `run_in_actor()` off the ria nursery
+- `tractor/spawn/_trio.py` — reaper branch + import drop
+- `tractor/spawn/_mp.py` — same as `_trio.py`
+- `ai/conc-anal/ria_nursery_removal_plan.md` — plan +
+  probe history
+
+## Human edits
+
+None yet — committed via the drafted
+`.claude/git_commit_msg_ria_step_a.md` (user-driven
+`git commit --edit`).

ai/prompt-io/claude/20260702T165806Z_a34aaf98_prompt_io.raw.md

@@ -0,0 +1,55 @@
+---
+model: claude-fable-5
+service: claude
+timestamp: 2026-07-02T16:58:06Z
+git_ref: a34aaf98
+diff_cmd: git diff a34aaf98..wkt/to_actor_subpkg
+---
+
+# Raw AI output (diff-ref mode)
+
+Step-A code is committed on `wkt/to_actor_subpkg`
+directly after `a34aaf98`; per diff-ref mode the verbatim
+content is reachable via the pointers below.
+
+## Generated files
+
+> `git diff a34aaf98..wkt/to_actor_subpkg -- tractor/runtime/_supervise.py`
+
+New `_reap_ria_portals(an, errors, ria_children=None)`
+helper (one `_spawn.cancel_on_completion()` task per ria
+child under `collapse_eg()` + a local nursery);
+`run_in_actor()` drops `nursery=self._ria_nursery`; happy
+path awaits the reap right after `._join_procs.set()`;
+inner error handler snapshots ria pairs, runs
+`await an.cancel()` then a `move_on_after(0.5)`-bounded
+reap over the snapshot.
+
+> `git diff a34aaf98..wkt/to_actor_subpkg -- tractor/spawn/_trio.py`
+> `git diff a34aaf98..wkt/to_actor_subpkg -- tractor/spawn/_mp.py`
+
+Both backends: the post-`_join_procs` block collapses to
+a bare `soft_kill()` (membership branch, per-child reaper
+nursery, reaper-cancel logging and the now-unused
+`cancel_on_completion` imports all removed).
+
+> `git diff a34aaf98..wkt/to_actor_subpkg -- ai/conc-anal/ria_nursery_removal_plan.md`
+
+Agent-verified machinery map, 3-step design (A/B/C),
+probe history (deadlock -> sequencing fix -> bound
+tightening) and risk register.
+
+## Test runs (verbatim)
+
+```
+tests/test_cancellation.py: 20 passed, 1 xfailed in 77.28s
+tests/test_to_actor.py + tests/test_spawning.py: 20 passed
+full-suite (SIGINT at 540s bound, ~30s short):
+303 passed, 9 skipped, 1 xfailed, 2 xpassed in 539.42s
+```
+
+Failed intermediates (not in final diff): concurrent
+reap+`an.cancel()` deadlocked `test_multierror`; a 3s
+reap bound blew
+`test_cancel_while_childs_child_in_sync_sleep`'s
+`fail_after` deadline.

ai/prompt-io/prompts/issue_477.md

@@ -0,0 +1,7 @@
+NOTE: you MUST pause this work at 12:50PM EST (BEFORE your weekly
+limit reset) for review by a human!
+
+---
+
+attempt to resolve https://github.com/goodboy/tractor/issues/477
+do it with /open-wkt.

examples/parallelism/to_actor_one_shots.py

@@ -0,0 +1,83 @@
+'''
+`tractor.to_actor.run()`: one-shot single-task subactor
+invocation, the SC-parallelism sibling of
+`trio.to_thread.run_sync()` (and `anyio.to_process`).
+
+Each call spawns a subactor, schedules the async fn as
+its lone remote task, waits on the result and reaps the
+subactor. Concurrency composes the plain `trio` way:
+schedule multiple one-shot calls in a local task nursery
+against a shared actor-nursery; any remote error raises
+directly in the task which scheduled it.
+
+'''
+import math
+
+import tractor
+import trio
+
+
+async def is_prime(
+    n: int,
+) -> bool:
+    if n < 2:
+        return False
+    if n == 2:
+        return True
+    if n % 2 == 0:
+        return False
+
+    sqrt_n = int(math.floor(math.sqrt(n)))
+    for i in range(3, sqrt_n + 1, 2):
+        if n % i == 0:
+            return False
+    return True
+
+
+async def main() -> None:
+
+    # fully implicit one-shot: boots the actor-runtime,
+    # spawns a subactor, runs the task, reaps the
+    # subactor, tears the runtime back down.
+    assert await tractor.to_actor.run(
+        is_prime,
+        n=2,
+    )
+
+    # the "worker-pool-ish" pattern from the original
+    # `concurrent.futures` example: one subactor per
+    # input, all concurrent, results and errors
+    # collected by caller-side tasks.
+    results: dict[int, bool] = {}
+
+    async def check(
+        an: tractor.ActorNursery,
+        n: int,
+        i: int,
+    ) -> None:
+        results[n] = await tractor.to_actor.run(
+            is_prime,
+            an=an,
+            name=f'prime_checker_{i}',
+            n=n,
+        )
+
+    inputs: list[int] = [
+        7,
+        8,
+        3691,
+        3693,
+    ]
+    async with (
+        tractor.open_nursery() as an,
+        trio.open_nursery() as tn,
+    ):
+        for i, n in enumerate(inputs):
+            tn.start_soon(check, an, n, i)
+
+    for n, prime in sorted(results.items()):
+        print(f'{n} is prime: {prime}')
+
+
+if __name__ == '__main__':
+    trio.run(main)

tests/test_to_actor.py

@@ -0,0 +1,271 @@
+'''
+`tractor.to_actor`: one-shot single-remote-task API suite.
+
+Verifies the "spiritual successor" to (and eventual
+replacement of) `ActorNursery.run_in_actor()`; see
+https://github.com/goodboy/tractor/issues/477
+
+'''
+from functools import partial
+
+import pytest
+import trio
+import tractor
+from tractor import (
+    RemoteActorError,
+    to_actor,
+)
+from tractor._testing import tractor_test
+
+
+async def add_one(
+    n: int,
+) -> int:
+    return n + 1
+
+
+async def raise_value_error() -> None:
+    raise ValueError('kaboom')
+
+
+@tractor_test
+async def test_one_shot_in_private_nursery(
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    No `an`/`portal` provided: a private actor-nursery
+    is opened (and torn down) scoped to just the call.
+
+    '''
+    assert await to_actor.run(
+        add_one,
+        n=1,
+    ) == 2
+
+
+def test_one_shot_boots_implicit_runtime(
+    reg_addr: tuple,
+    start_method: str,
+    loglevel: str,
+):
+    '''
+    Outside any actor-runtime `to_actor.run()` boots one
+    implicitly (just like bare `open_nursery()` usage)
+    configured via pass-through `runtime_kwargs`.
+
+    '''
+    async def main() -> None:
+        assert tractor.current_actor(
+            err_on_no_runtime=False,
+        ) is None
+        result = await to_actor.run(
+            add_one,
+            n=41,
+            runtime_kwargs=dict(
+                registry_addrs=[reg_addr],
+                start_method=start_method,
+                loglevel=loglevel,
+            ),
+        )
+        assert result == 42
+
+    trio.run(main)
+
+
+@tractor_test
+async def test_remote_error_relayed_to_caller_task(
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    A remote task error is raised directly in the
+    caller's task as a boxed `RemoteActorError` instead
+    of surfacing at actor-nursery teardown as with the
+    legacy `.run_in_actor()` API.
+
+    '''
+    with pytest.raises(RemoteActorError) as excinfo:
+        await to_actor.run(raise_value_error)
+
+    assert excinfo.value.boxed_type is ValueError
+
+
+@tractor_test
+async def test_spawn_from_caller_nursery(
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    Pass a caller-managed `an: ActorNursery` for the
+    spawn; the subactor is still one-shot reaped by the
+    time the call returns.
+
+    '''
+    async with tractor.open_nursery() as an:
+        assert await to_actor.run(
+            add_one,
+            an=an,
+            n=10,
+        ) == 11
+
+
+@tractor_test
+async def test_remote_error_from_caller_nursery(
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    With a caller-managed `an` the remote error also
+    surfaces in the caller's task, INSIDE the nursery
+    block, allowing inline (supervision-style) handling.
+
+    '''
+    async with tractor.open_nursery() as an:
+        with pytest.raises(RemoteActorError) as excinfo:
+            await to_actor.run(
+                raise_value_error,
+                an=an,
+            )
+
+        assert excinfo.value.boxed_type is ValueError
+
+
+@tractor_test
+async def test_reuse_existing_actor_via_portal(
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    Pass `portal=` to schedule the one-shot task in an
+    already-running actor; no spawn, no implicit reap.
+
+    '''
+    async with tractor.open_nursery() as an:
+        portal: tractor.Portal = await an.start_actor(
+            'one_shot_worker',
+            enable_modules=[__name__],
+        )
+        for i in range(3):
+            assert await to_actor.run(
+                add_one,
+                portal=portal,
+                n=i,
+            ) == i + 1
+
+        # still alive: caller owns the actor's lifetime.
+        await portal.cancel_actor()
+
+
+@tractor_test
+async def test_concurrent_one_shots_from_task_nursery(
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    The worker-pool-ish pattern from #477: concurrency
+    is composed with a plain (caller-side) `trio` task
+    nursery scheduling multiple one-shot calls against
+    a shared caller-managed actor-nursery; error
+    collection thus lives entirely in caller-code.
+
+    '''
+    results: dict[int, int] = {}
+
+    async def one_shot(
+        an: tractor.ActorNursery,
+        i: int,
+    ) -> None:
+        results[i] = await to_actor.run(
+            add_one,
+            an=an,
+            name=f'one_shot_{i}',
+            n=i,
+        )
+
+    async with (
+        tractor.open_nursery() as an,
+        trio.open_nursery() as tn,
+    ):
+        for i in range(4):
+            tn.start_soon(one_shot, an, i)
+
+    assert results == {
+        i: i + 1 for i in range(4)
+    }
+
+
+def test_rejects_sync_fn():
+    '''
+    Non-async callables error BEFORE any spawn (or even
+    runtime-boot) happens.
+
+    '''
+    def not_async() -> None:
+        ...
+
+    with pytest.raises(TypeError):
+        trio.run(
+            partial(
+                to_actor.run,
+                not_async,
+            )
+        )
+
+
+def test_rejects_streaming_fn():
+    '''
+    Async-gen (streaming) fns are not one-shot-able,
+    same constraint as `Portal.run()`.
+
+    '''
+    async def agen():
+        yield 1
+
+    with pytest.raises(TypeError):
+        trio.run(
+            partial(
+                to_actor.run,
+                agen,
+            )
+        )
+
+
+def test_rejects_portal_and_an_combo():
+    '''
+    `portal=` and `an=` are mutually exclusive
+    placement options.
+
+    '''
+    with pytest.raises(ValueError):
+        trio.run(
+            partial(
+                to_actor.run,
+                add_one,
+                portal=object(),
+                an=object(),
+                n=1,
+            )
+        )
+
+
+def test_rejects_runtime_kwargs_with_placement():
+    '''
+    `runtime_kwargs` only applies when the call opens
+    its own private actor-nursery; passing it alongside
+    a placement opt is an error, never silently
+    ignored.
+
+    '''
+    with pytest.raises(ValueError):
+        trio.run(
+            partial(
+                to_actor.run,
+                add_one,
+                an=object(),
+                runtime_kwargs=dict(
+                    loglevel='cancel',
+                ),
+                n=1,
+            )
+        )

tractor/__init__.py

@@ -62,6 +62,7 @@ from .devx import (
     post_mortem as post_mortem,
 )
 from . import msg as msg
+from . import to_actor as to_actor
 from ._root import (
     run_daemon as run_daemon,
     open_root_actor as open_root_actor,

tractor/runtime/_supervise.py

@@ -383,12 +383,13 @@ class ActorNursery:
         )
 
     # TODO: DEPRECATE THIS:
-    # -[ ] impl instead as a hilevel wrapper on
-    #   top of a `@context` style invocation.
-    #  |_ dynamic @context decoration on child side
-    #  |_ implicit `Portal.open_context() as (ctx, first):`
-    #    and `return first` on parent side.
-    #  |_ mention how it's similar to `trio-parallel` API?
+    # -[x] impl instead as a hilevel wrapper on top of
+    #   the lower level daemon-spawn + portal APIs
+    #  |_ see `.to_actor.run()` (issue #477) which does
+    #    `.start_actor()` + `Portal.run()` + a one-shot
+    #    reap via `Portal.cancel_actor()`.
+    # -[ ] emit a `DeprecationWarning` here (requires
+    #   migrating all in-repo usage first!)
     # -[ ] use @api_frame on the wrapper
     async def run_in_actor(
         self,
@@ -416,6 +417,11 @@ class ActorNursery:
         until the task spawned by executing ``fn`` completes at which point
         the actor is terminated.
 
+        NOTE: prefer the (eventual) replacement API
+        `tractor.to_actor.run()` which delivers the same
+        one-shot semantics decoupled from this nursery's
+        internal spawn machinery; see issue #477.
+
         '''
         __runtimeframe__: int = 1  # noqa
         mod_path: str = fn.__module__
@@ -432,8 +438,6 @@ class ActorNursery:
             ),
             bind_addrs=bind_addrs,
             loglevel=loglevel,
-            # use the run_in_actor nursery
-            nursery=self._ria_nursery,
             infect_asyncio=infect_asyncio,
             inherit_parent_main=inherit_parent_main,
             proc_kwargs=proc_kwargs
@@ -574,6 +578,51 @@ class ActorNursery:
         self._join_procs.set()
 
 
+async def _reap_ria_portals(
+    an: ActorNursery,
+    errors: dict[tuple[str, str], BaseException],
+    ria_children: list[tuple[Portal, Actor]]|None = None,
+) -> None:
+    '''
+    Wait on and stash the final result/error from every
+    `.run_in_actor()`-spawned child then cancel its actor
+    runtime, one `_spawn.cancel_on_completion()` task per
+    child.
+
+    Replaces the per-child reaper task formerly spawned by
+    the spawn backends (keyed off
+    `._cancel_after_result_on_exit` membership) which
+    required routing such children into the (now removable)
+    `._ria_nursery`. Only call AFTER `._join_procs` is set
+    so user code inside the nursery block retains exclusive
+    result-await access; see the "manually await results"
+    note in `spawn._mp.mp_proc()`.
+
+    '''
+    if ria_children is None:
+        ria_children: list[tuple[Portal, Actor]] = [
+            (portal, subactor)
+            for subactor, _, portal in an._children.values()
+            if portal in an._cancel_after_result_on_exit
+        ]
+    if not ria_children:
+        return
+
+    async with (
+        collapse_eg(),
+        trio.open_nursery() as tn,
+    ):
+        portal: Portal
+        subactor: Actor
+        for portal, subactor in ria_children:
+            tn.start_soon(
+                _spawn.cancel_on_completion,
+                portal,
+                subactor,
+                errors,
+            )
+
+
 @acm
 async def _open_and_supervise_one_cancels_all_nursery(
     actor: Actor,
@@ -635,6 +684,11 @@ async def _open_and_supervise_one_cancels_all_nursery(
                     )
                     an._join_procs.set()
 
+                    # collect results (and errors) from all
+                    # `.run_in_actor()` children then cancel
+                    # each, one reaper task per child.
+                    await _reap_ria_portals(an, errors)
+
                 except BaseException as _inner_err:
                     inner_err = _inner_err
                     errors[actor.aid.uid] = inner_err
@@ -698,9 +752,39 @@ async def _open_and_supervise_one_cancels_all_nursery(
                                 # '------ - ------'
                             )
 
+                        # snapshot `.run_in_actor()` children
+                        # BEFORE cancelling: each backend
+                        # spawn-task pops its `._children`
+                        # entry as the proc gets reaped.
+                        ria_children: list = [
+                            (portal, subactor)
+                            for subactor, _, portal
+                            in an._children.values()
+                            if portal in
+                            an._cancel_after_result_on_exit
+                        ]
                         # cancel all subactors
                         await an.cancel()
 
+                        # then collect any already-relayed
+                        # results/errors from ria children.
+                        # Tightly bounded: anything
+                        # collectable is already queued in
+                        # the local ctx (relayed BEFORE the
+                        # cancel above); a child hard-killed
+                        # without relaying just parks its
+                        # reaper which then self-cleans (a
+                        # `trio.Cancelled` result is never
+                        # stashed), mirroring the old
+                        # backend-side reaper-vs-`soft_kill`
+                        # cancel race.
+                        with trio.move_on_after(0.5):
+                            await _reap_ria_portals(
+                                an,
+                                errors,
+                                ria_children=ria_children,
+                            )
+
             # ria_nursery scope end
 
         # TODO: this is the handler around the ``.run_in_actor()``

tractor/spawn/_mp.py

@@ -48,7 +48,6 @@ from ._entry import _mp_main
 # by `try_set_start_method()` after module load time.
 from . import _spawn
 from ._spawn import (
-    cancel_on_completion,
     proc_waiter,
     soft_kill,
 )
@@ -187,31 +186,17 @@ async def mp_proc(
         with trio.CancelScope(shield=True):
             await actor_nursery._join_procs.wait()
 
-        async with trio.open_nursery() as nursery:
-            if portal in actor_nursery._cancel_after_result_on_exit:
-                nursery.start_soon(
-                    cancel_on_completion,
-                    portal,
-                    subactor,
-                    errors
-                )
-
         # This is a "soft" (cancellable) join/reap which
         # will remote cancel the actor on a ``trio.Cancelled``
-            # condition.
+        # condition. Any `.run_in_actor()` result-reaping
+        # happens up in the `ActorNursery` machinery (see
+        # `_supervise._reap_ria_portals()`), NOT here.
         await soft_kill(
             proc,
             proc_waiter,
             portal
         )
 
-            # cancel result waiter that may have been spawned in
-            # tandem if not done already
-            log.warning(
-                "Cancelling existing result waiter task for "
-                f"{subactor.aid.uid}")
-            nursery.cancel_scope.cancel()
-
     finally:
         # hard reap sequence
         if proc.is_alive():

tractor/spawn/_trio.py

@@ -50,7 +50,6 @@ from tractor.msg import (
     pretty_struct,
 )
 from ._spawn import (
-    cancel_on_completion,
     hard_kill,
     soft_kill,
 )
@@ -195,32 +194,17 @@ async def trio_proc(
         with trio.CancelScope(shield=True):
             await actor_nursery._join_procs.wait()
 
-        async with trio.open_nursery() as nursery:
-            if portal in actor_nursery._cancel_after_result_on_exit:
-                nursery.start_soon(
-                    cancel_on_completion,
-                    portal,
-                    subactor,
-                    errors
-                )
-
         # This is a "soft" (cancellable) join/reap which
         # will remote cancel the actor on a ``trio.Cancelled``
-            # condition.
+        # condition. Any `.run_in_actor()` result-reaping
+        # happens up in the `ActorNursery` machinery (see
+        # `_supervise._reap_ria_portals()`), NOT here.
         await soft_kill(
             proc,
             trio.Process.wait,  # XXX, uses `pidfd_open()` below.
             portal
         )
 
-            # cancel result waiter that may have been spawned in
-            # tandem if not done already
-            log.cancel(
-                'Cancelling portal result reaper task\n'
-                f'c)> {subactor.aid.reprol()!r}\n'
-            )
-            nursery.cancel_scope.cancel()
-
     finally:
         # XXX NOTE XXX: The "hard" reap since no actor zombies are
         # allowed! Do this **after** cancellation/teardown to avoid

tractor/to_actor/__init__.py

@@ -0,0 +1,33 @@
+# tractor: distributed structured concurrency.
+# Copyright 2018-eternity Tyler Goodlet.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+`tractor.to_actor`: high-level "one-shot" remote-task APIs.
+
+Adopts the "run it over there" parlance from analogous
+(sibling-library) APIs like `trio.to_thread` and
+`anyio.to_process` but for SC-supervised actors: spawn (or
+reuse) a subactor, schedule a single remote task, wait on
+its result and (when the call owns the subactor) reap it.
+
+The "spiritual successor" to (and eventual replacement of)
+the `ActorNursery.run_in_actor()` API; see
+https://github.com/goodboy/tractor/issues/477
+
+'''
+from ._api import (
+    run as run,
+)

tractor/to_actor/_api.py

@@ -0,0 +1,226 @@
+# tractor: distributed structured concurrency.
+# Copyright 2018-eternity Tyler Goodlet.
+
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+'''
+One-shot remote-task invocation built on spawn-and-portal
+primitives.
+
+Implemented (as prescribed by #477) entirely "on top of"
+the lower level daemon-actor spawn + portal APIs,
+
+- `ActorNursery.start_actor()` for (daemon-style) subactor
+  spawning,
+- `Portal.run()` for scheduling the lone remote task and
+  waiting on its result,
+- `Portal.cancel_actor()` for reaping the subactor once
+  that result (or error) arrives,
+
+such that error collection and propagation happens in the
+*caller's task* (and thus whatever `trio` nursery/scope
+encloses it) instead of inside the actor-nursery's
+spawn-machinery nurseries as with the (to be deprecated)
+`ActorNursery.run_in_actor()` API.
+
+'''
+from __future__ import annotations
+import inspect
+from typing import (
+    Any,
+    Callable,
+    TYPE_CHECKING,
+)
+
+from ..runtime._supervise import (
+    ActorNursery,
+    open_nursery,
+)
+
+if TYPE_CHECKING:
+    from ..discovery._addr import UnwrappedAddress
+    from ..runtime._portal import Portal
+
+
+def _validate_one_shot_fn(
+    fn: Callable,
+) -> None:
+    '''
+    Ensure `fn` is a non-streaming async function, raise
+    a `TypeError` otherwise.
+
+    The same constraint enforced by `Portal.run()` but
+    checked up-front, BEFORE any subactor is spawned.
+
+    '''
+    if not (
+        inspect.iscoroutinefunction(fn)
+        and
+        not getattr(
+            fn,
+            '_tractor_stream_function',
+            False,
+        )
+    ):
+        raise TypeError(
+            f'{fn!r} must be a non-streaming async '
+            f'function!'
+        )
+
+
+async def _invoke_in_subactor(
+    an: ActorNursery,
+    fn: Callable,
+    name: str,
+    spawn_kwargs: dict[str, Any],
+    fn_kwargs: dict[str, Any],
+) -> Any:
+    '''
+    Spawn a (daemon) subactor via `an.start_actor()`,
+    schedule `fn` as its lone remote task via
+    `Portal.run()` and, ALWAYS, reap the subactor once
+    that task's result (or error) has been delivered.
+
+    '''
+    portal: Portal = await an.start_actor(
+        name,
+        **spawn_kwargs,
+    )
+    try:
+        return await portal.run(
+            fn,
+            **fn_kwargs,
+        )
+    finally:
+        # one-shot semantics: the subactor's lifetime is
+        # bound to its lone task's completion; the
+        # cancel-req's bounded wait is shielded
+        # internally (see `Portal.cancel_actor()`) so
+        # this reap also runs when the caller's scope
+        # was itself cancelled.
+        await portal.cancel_actor()
+
+
+async def run(
+    fn: Callable,
+    *,
+
+    # actor "placement": reuse an already-running peer
+    # via its `portal`, spawn a fresh subactor from
+    # a caller-managed `an: ActorNursery`, or, when
+    # neither is provided, open a private actor-nursery
+    # (implicitly booting the actor-runtime as needed)
+    # scoped to just this call.
+    portal: Portal|None = None,
+    an: ActorNursery|None = None,
+
+    # subactor spawn opts passed (mostly) verbatim to
+    # `ActorNursery.start_actor()`; unused when `portal`
+    # is provided.
+    name: str|None = None,
+    bind_addrs: list[UnwrappedAddress]|None = None,
+    enable_modules: list[str]|None = None,
+    loglevel: str|None = None,
+    debug_mode: bool|None = None,
+    infect_asyncio: bool = False,
+    inherit_parent_main: bool = True,
+    proc_kwargs: dict[str, Any]|None = None,
+
+    # passed verbatim to the private `open_nursery()`
+    # (and in turn any implicit `open_root_actor()`)
+    # when NO `an`/`portal` is provided.
+    runtime_kwargs: dict[str, Any]|None = None,
+
+    **fn_kwargs,  # explicit (keyword) args to `fn`
+
+) -> Any:
+    '''
+    Run the async `fn` as the lone task in a (new)
+    subactor, block waiting on its result and return it;
+    the distributed-parallelism equivalent of
+    `trio.to_thread.run_sync()`.
+
+    Unlike `ActorNursery.run_in_actor()` (which returns
+    a `Portal` whose result is only collected at
+    actor-nursery teardown) this is a plain "call and
+    wait" primitive: any remote error is raised HERE, in
+    the caller's task. Concurrency is composed the usual
+    `trio` way by scheduling multiple `run()` calls in
+    a local task nursery, ideally against a shared
+    caller-managed `an: ActorNursery` (see the test
+    suite for the canonical worker-pool-ish pattern).
+
+    '''
+    __runtimeframe__: int = 1  # noqa
+    _validate_one_shot_fn(fn)
+
+    if (
+        runtime_kwargs
+        and
+        (
+            an is not None
+            or
+            portal is not None
+        )
+    ):
+        raise ValueError(
+            '`runtime_kwargs` only applies when this '
+            'call opens its own private actor-nursery '
+            '(no `an`/`portal` provided)!'
+        )
+
+    if portal is not None:
+        if an is not None:
+            raise ValueError(
+                'Pass at most ONE of `portal` or `an`, '
+                'not both!'
+            )
+        return await portal.run(
+            fn,
+            **fn_kwargs,
+        )
+
+    name: str = name or fn.__name__
+    spawn_kwargs: dict[str, Any] = dict(
+        enable_modules=(
+            [fn.__module__]
+            +
+            (enable_modules or [])
+        ),
+        bind_addrs=bind_addrs,
+        loglevel=loglevel,
+        debug_mode=debug_mode,
+        infect_asyncio=infect_asyncio,
+        inherit_parent_main=inherit_parent_main,
+        proc_kwargs=proc_kwargs,
+    )
+    if an is not None:
+        return await _invoke_in_subactor(
+            an,
+            fn,
+            name,
+            spawn_kwargs,
+            fn_kwargs,
+        )
+
+    async with open_nursery(
+        **(runtime_kwargs or {}),
+    ) as an:
+        return await _invoke_in_subactor(
+            an,
+            fn,
+            name,
+            spawn_kwargs,
+            fn_kwargs,
+        )