Clean broadcast cancellation diagnostics

Bound `BroadcastState.cancelled` entries to receiver progress,
terminal state and resource lifetime instead of retaining completed
`Task`s indefinitely.

Deats,
- make EOC durable so awakened peers never re-enter a closed source.
- close root broadcasters during explicit `MsgStream` and
  `LinkedTaskChannel` teardown without re-entrant EOC closure or
  breaking `MsgStream.aclose()` overrides.
- reject non-positive fan-out retention capacity before constructing
  an unusable zero-length queue.
- cover child/root cancellation cleanup, terminal peer wakeups,
  wrapper teardown, subclass compatibility and zero-buffer rejection.

Prompt-IO: ai/prompt-io/opencode/20260813T181901Z_a2e0df4b_prompt_io.md

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

ai/conc-anal/ria_nursery_removal_plan.md

@@ -0,0 +1,463 @@
+# `_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.
+
+## Step-B outcome (2026-07-02, done in tree)
+
+Step A landed as `5cd190c5` (code) + `99310269` (docs).
+Step B implemented on top (uncommitted):
+
+- `._ria_nursery` is GONE — the inner
+  `async with (collapse_eg(), trio.open_nursery() as
+  ria_nursery)` layer in
+  `_open_and_supervise_one_cancels_all_nursery` is deleted;
+  `da_nursery` is now the single nursery for ALL subactors.
+- `ActorNursery.__init__` drops the `ria_nursery` param +
+  the `self._ria_nursery` attr; `start_actor()` drops its
+  `nursery=` escape-hatch param (uses `self._da_nursery`
+  directly).
+- `._cancel_after_result_on_exit` STAYS — it's the
+  ria-child discriminator for `_reap_ria_portals()`.
+
+Deliberately NOT done (deferred to its own higher-risk PR,
+flagged with a TODO at the outer `except`): merging the two
+error handlers into one. Rationale — collapsing the empty
+nursery is provably behavior-preserving (a zero-task
+`trio.open_nursery()` only adds a checkpoint), whereas the
+inner `except BaseException` (swallow-into-`errors`) and
+outer `except (...)` (re-raise, safety-net for the inner
+handler's own non-shielded awaits) have DIFFERENT
+semantics; merging changes error/cancel propagation and
+wants isolated review + its own gate. Both handlers are
+kept, now nested directly under the single nursery.
+
+Why the collapse is safe: post-step-A NOTHING spawns into
+`ria_nursery` (its only reader, `run_in_actor`'s
+`nursery=self._ria_nursery`, was removed in A; the stored
+attr was never read again). So the layer was pure dead
+weight.
+
+Gate (trio backend, all 0-failure):
+- targeted set (`test_cancellation test_spawning test_local
+  test_rpc test_to_actor`) = 49 passed, 1 xfailed.
+- tail set (`test_reg_err_types remote_exc_relay
+  resource_cache ringbuf root_infect_asyncio root_runtime
+  runtime shm task_broadcasting trioisms trionics/`) = 63
+  passed, 1 skipped, 5 xfailed.
+- full-suite head ~73% (subdirs + `test_2way`..`test_pubsub`)
+  = 303 passed before the known-flaky `test_dynamic_pub_sub`
+  TooSlowError stall (pre-existing; same hang in the step-A
+  full run). Suite ran slow this session (~13min vs 555s
+  cold, likely thermal from back-to-back runs), never
+  completing within an 800s bound — but split across the
+  above three runs EVERY module passed under step B.
+
+## Step-B2 outcome (2026-07-02, done in tree)
+
+Step B committed as `9201a2ed` (code) + `d2e812fb` (docs), then
+branched to `drop_ria_nursery`. Step B2 (the deferred
+handler-merge) implemented on top (uncommitted):
+
+- the two nested handlers in
+  `_open_and_supervise_one_cancels_all_nursery` collapse to
+  ONE `except BaseException as _scope_err` + the existing
+  `finally`. The `outer_err`/`inner_err` locals go away.
+
+Why it's safe (trace, not hope): the OLD inner handler records
+`errors[actor.aid.uid]` as its FIRST statement (before any
+await). So whenever an error path runs, `errors` is non-empty.
+The OLD outer handler was only reachable via leakage from the
+inner handler (it catches `BaseException`, so nothing from the
+`yield` scope bypasses it) — and by then `errors` is already
+populated, so the `finally`'s `raise` from `errors` ALWAYS
+superseded the outer handler's own `raise`. i.e. the outer
+`raise` was dead. The outer handler's other effects
+(`_scope_error`, a 2nd debugger-wait, child-cancel) are
+redundant with the merged handler + `finally`. So one handler
++ `finally` is observably equivalent.
+
+Residual nuance (accepted): in the rare "`trio.Cancelled`
+delivered during the non-shielded `maybe_wait_for_debugger`"
+path, the merged form may leave `_cancel_called` False (cancel
+happens after the wait), so `open_nursery`'s tb-hiding guard
+(`not cancel_called and _scope_error`) can show a tb it
+previously hid. More informative, not less; no test asserts on
+it.
+
+Gate (box ran ~2.7x slow this session, load-induced
+`TooSlowError` flakiness on timing tests — NOT code; see
+[[env_cpu_throttle_masquerades_as_regression]]):
+- baseline (pre-B2 tip `9201a2ed`) full suite
+  (`-k 'not dynamic_pub_sub'`) = 300 passed + 1
+  `test_ext_types_over_ipc` `TooSlowError` that passes 6/6 in
+  isolation (4.89s).
+- B2 error/cancel gate (`test_cancellation remote_exc_relay
+  inter_peer_cancellation advanced_faults oob_cancellation
+  to_actor spawning local rpc`) = 71 passed, 1 xfailed
+  (125s).
+- B2 full-suite run: see `b2_full.log` (result appended on
+  completion). RECOMMEND a clean full-suite run on a
+  normal-speed box before this merges.
+
+## Regression + fix: ria-reap hang (2026-07-02)
+
+Human hit a full-suite hang on
+`test_infected_asyncio.py::test_tractor_cancels_aio`. Bisected:
+passes at pre-ria `a34aaf98` (0.59s), hangs at B2 `e617b498`
+(90s+). Root-caused to the STEP-A reaper hoist (`5cd190c5`),
+NOT B2 (`_reap_ria_portals` is byte-identical A->B2).
+
+Bug: the test does `run_in_actor(asyncio_actor)` then a USER
+`portal.cancel_actor()` and exits the block cleanly -> the
+happy path's `await _reap_ria_portals()`, which waits UNBOUNDED
+on `cancel_on_completion -> wait_for_result()`. The child was
+cancelled out-of-band so no final result is relayed -> parked
+forever. The OLD spawn-backend reaper was raced against
+`soft_kill()` (per-child nursery `cancel_scope.cancel()` on
+subproc death); the hoist dropped that race.
+
+Fix: `_reap_ria_portals()` runs each `cancel_on_completion()`
+in a local nursery alongside a `proc.poll()` death-watch that
+cancels the parked reaper once the subproc exits — restoring
+the old race, backend-agnostic (guarded by
+`hasattr(proc, 'poll')` for a future `subint` handle).
+
+Why POLL (`proc.poll()`) not the event-driven `wait_func`:
+the mp waiter (`_spawn.proc_waiter`) does
+`wait_readable(proc.sentinel)`, and `soft_kill()` is ALREADY
+awaiting that same fd concurrently in the daemon nursery — a
+2nd `wait_readable` on one fd raises `trio.BusyResourceError`.
+(`trio.Process.wait()` IS multi-waiter-safe, but mp has no
+async equivalent.) `proc.poll()` — the same liveness check
+`soft_kill` itself falls back to — is the conflict-free common
+denominator. Verified: poll-fix passes on BOTH trio and
+mp_spawn.
+
+Also added a per-test anti-hang guard: wrapped
+`test_tractor_cancels_aio`'s `main()` in
+`with trio.fail_after(9 * cpu_perf_headroom())` — the blessed
+pattern (`pytest-timeout`'s global cap is intentionally off;
+breaks trio under fork backends, see `pyproject` NOTE). So a
+future recurrence FAILS FAST instead of hanging the suite.
+(Several other tests in the file are still guardless —
+`test_aio_simple_error`, `test_trio_error_cancels_intertask_chan`,
+`test_aio_errors_and_channel_propagates_and_closes` — candidate
+follow-up sweep.)
+
+Lesson: the B2 focused gate OMITTED `test_infected_asyncio`
+(and the full runs were clipped/slow), so the step-A hang
+slipped through. Any future ria-touching change MUST gate
+`test_infected_asyncio` explicitly.
+
+Gate: `test_tractor_cancels_aio` green (trio 1.53s, mp 3.98s);
+fix gate (`test_infected_asyncio test_cancellation test_to_actor
+test_spawning`) = 74 passed, 3 xfailed, 0 failures.
+
+## PAUSED (2026-07-02): re-assess the reaper's SCOPE
+
+User's insight (compelling — likely the real root cause of
+the hang, not just the missing proc-death race):
+
+> the "hoisting" of 5cd190c5 was just not really done right
+> — the hoist should have been into the `to_actor` scope,
+> not `_supervise`.
+
+The argument: `.run_in_actor()`'s result-waiting/reaping got
+hoisted into `_supervise._reap_ria_portals` (nursery-machinery
+scope), which has NO natural cancel-scope to bound a parked
+`wait_for_result()` — hence the awkward proc-death race +
+the poll-vs-`proc_waiter` dilemma. If the result-wait instead
+lived in the `to_actor` one-shot scope
+(`to_actor._invoke_in_subactor()`), it would sit right next to
+the caller's `an` + a local `trio` task-nursery + cancel-scope
+(the `trio.to_thread`-style model #477 actually wants) — so
+bounding/cancelling the wait is trivial and the hang
+dissolves from correct scoping rather than a bolt-on race.
+
+Follow-on to re-evaluate on resume:
+- should `_reap_ria_portals` exist AT ALL, or should
+  result-waiting move entirely into
+  `to_actor._invoke_in_subactor()`?
+- reimplement legacy `run_in_actor()` on top of
+  `to_actor.run()` so `_reap_ria_portals` +
+  `_cancel_after_result_on_exit` can be DROPPED from
+  `_supervise` entirely (the true #477 simplification)?
+- the poll-vs-event decision is MOOT under this re-scoping.
+
+State at pause: `test_infected_asyncio` anti-hang guard
+COMMITTED (`d1fb4a1a`, intentionally red w/o the fix — the
+user's failing-test-first convention). The poll-based reap
+fix in `_supervise.py` is UNCOMMITTED and likely SUPERSEDED
+by the re-scoping — do NOT land it as-is.
+
+## RESOLVED (2026-07-06): migrate everything, remove the API
+
+The PAUSED re-assessment concluded decisively: rather than
+re-scope `_reap_ria_portals` (or bolt any hack onto it), the
+`run_in_actor()` API itself was REMOVED — its non-blocking
+"result at teardown" semantic predates streaming and confused
+more than it served. Every in-repo caller was migrated
+per-file/-group (each its own commit, each gated):
+
+- tests: `test_infected_asyncio` `test_runtime` `test_rpc`
+  `test_spawning` `test_pubsub` `test_registrar`
+  `test_cancellation` (3 groups) `test_advanced_streaming`.
+- examples: 4 non-debugging + all 8 `debugging/` REPL scripts
+  (debugger suite byte-identical green, 28p/6s).
+- docs: 8 rst pages + the `experimental/_pubsub` docstring.
+
+Migration patterns (the `run_in_actor` shape -> successor):
+
+- blocking result        -> `to_actor.run(fn, an=an, ...)`
+- fire-&-forget/forever  -> bg `to_actor.run()` task in a local
+                            `trio` task-nursery (or `start_actor`
+                            + bg `Portal.run()` when a portal
+                            handle is needed)
+- concurrent fan-out     -> N bg `to_actor.run()` tasks / or
+                            `gather_contexts([p.open_context(..)])`
+- reap-all-error-collect -> the "collect don't cancel" pattern:
+                            each one-shot catches + stashes its
+                            `RemoteActorError`, group raised
+                            after the task-nursery joins (see
+                            `examples/debugging/multi_subactors.py`)
+- mutual-rendezvous      -> peers must OUTLIVE both dialogs:
+                            `start_actor()` daemons + concurrent
+                            `Portal.run()`s + explicit
+                            `an.cancel()` (eager one-shot reap
+                            races the slower peer's dial of the
+                            winner's dead sockaddr; found via
+                            `test_trynamic_trio` flake).
+
+Semantic deltas (tests loosened accordingly):
+
+- teardown-reap-all BEG-of-N is GONE: local task-nurseries are
+  cancel-on-first, raced siblings' `Cancelled`s are absorbed,
+  and the runtime's `collapse_eg()` unwraps every single-member
+  group at each actor boundary — a fully-raced nested tree
+  relays a bare (annotated) `RemoteActorError` chain.
+- `test_multierror_fast_nursery`'s obsolete BEG-of-25 assertion
+  deleted; `test_concurrent_start_error_reaps_all` retains its
+  high-fan-out startup/cancel/reap stress under caller-scoped
+  semantics.
+- `test_nested_multierrors` re-purposed separately as deep-tree
+  cancel-cascade stress w/ a race-tolerant shape walk.
+
+Final excision (after zero callers remained): `run_in_actor()`,
+`._cancel_after_result_on_exit`, `_reap_ria_portals()`,
+`Portal._submit_for_result/._expect_result_ctx/
+.wait_for_result()/.result()`, `exhaust_portal()`,
+`cancel_on_completion()`, `NoResult` — net -402 lines. The
+reap-hang class (unbounded `wait_for_result` in machinery
+scope) dissolves structurally: the only result-wait left lives
+in the caller's task inside its own cancel-scope; the
+`d1fb4a1a` anti-hang guard test passes by construction. The
+poll-vs-`proc_waiter` debate is moot as predicted.
+
+## Follow-up sketch: `to_actor.open_one_shot()` (run-async parity)
+
+If deferred-result parity is ever wanted, the design that needs
+NO runtime coupling, NO returned `Portal` and NO cancel-relay
+`trio.Event` machinery:
+
+    async with to_actor.open_one_shot(
+        fn, an=an, **kws,
+    ) as one_shot:
+        ...                       # concurrent caller work
+        val = await one_shot.wait()   # optional; errors always
+                                      # propagate at scope exit
+
+an `@acm` that opens a private task-nursery, `start_soon`s ONE
+task running the existing blocking `run()` and stashes the
+value in a slot + sets a done-`trio.Event` (a memo, not a
+cancel relay). Cancellation = plain scope-cancel of the acm's
+nursery (the parked `Portal.run()` unwinds via `Cancelled`, the
+shielded `cancel_actor()` reap still runs); a child error
+raises into the acm scope so an un-`wait()`ed one-shot can
+never silently drop its error. i.e. the old reaper's job is
+done by scoping, not machinery. ~40 lines, all in
+`to_actor/_api.py`, zero `_supervise` involvement.
+
+## Verification gate
+
+- per-migration-commit module gates on `trio` (+ `mp_spawn`
+  spot-gates incl. `test_infected_asyncio` per the B2 lesson);
+  `tests/devx/test_debugger.py` for the REPL flows.
+- full suite on `trio` + `mp_spawn` at branch tip + CI matrix
+  via draft PR #484.

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/claude/20260702T172233Z_5cd190c5_prompt_io.md

@@ -0,0 +1,60 @@
+---
+model: claude-opus-4-8
+service: claude
+session: c7860302-6a33-4d92-821c-0807d68be464
+timestamp: 2026-07-02T17:22:33Z
+git_ref: 5cd190c5
+scope: code
+substantive: true
+raw_file: 20260702T172233Z_5cd190c5_prompt_io.raw.md
+---
+
+## Prompt
+
+Branched session (`to_actor_subpkg_POST_WEEKLY_CUTOFF_2OPUS`,
+now on Opus 4.8) continuing the `._ria_nursery` removal after
+step A committed + pushed (`5cd190c5`/`99310269`):
+
+> ok continue where your superior left off please
+
+then, after the commit-split discussion:
+
+> yup committed and pushed so take a look ma fren.
+
+i.e. green-lighting step-B prep against the committed step-A
+tree.
+
+## Response summary
+
+Step B of the `._ria_nursery` removal (issue #477): delete
+the now-vestigial secondary nursery,
+
+- collapse the inner `async with (collapse_eg(),
+  trio.open_nursery() as ria_nursery)` layer in
+  `_open_and_supervise_one_cancels_all_nursery` — `da_nursery`
+  becomes the single nursery for ALL subactors.
+- `ActorNursery.__init__` drops the `ria_nursery` param + the
+  `self._ria_nursery` attr; `start_actor()` drops its
+  `nursery=` escape-hatch param.
+- `._cancel_after_result_on_exit` kept (ria-child
+  discriminator for `_reap_ria_portals()`).
+
+Verified behavior-preserving via a first-principles argument
+(zero-task nursery = a bare checkpoint) + the targeted gate
+(`test_cancellation test_spawning test_local test_rpc
+test_to_actor` = 49 passed, 1 xfailed on trio). The two
+error handlers were deliberately NOT merged — that changes
+propagation semantics and is deferred to its own PR (TODO
+left at the outer `except`).
+
+## Files changed
+
+- `tractor/runtime/_supervise.py` — collapse the ria nursery
+  layer + drop the ctor/`start_actor` params + refresh the
+  now-stale nursery comments
+
+## Human edits
+
+None yet — committed via the drafted
+`.claude/git_commit_msg_ria_step_b.md` (user-driven
+`git commit --edit`).

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

@@ -0,0 +1,51 @@
+---
+model: claude-opus-4-8
+service: claude
+timestamp: 2026-07-02T17:22:33Z
+git_ref: 5cd190c5
+diff_cmd: git diff 5cd190c5..wkt/to_actor_subpkg
+---
+
+# Raw AI output (diff-ref mode)
+
+Step-B code lives on `wkt/to_actor_subpkg` after `5cd190c5`;
+per diff-ref mode the verbatim content is reachable via the
+pointer below.
+
+## Generated files
+
+> `git diff 5cd190c5..wkt/to_actor_subpkg -- tractor/runtime/_supervise.py`
+
+- `ActorNursery.__init__`: `ria_nursery` param removed;
+  `self._ria_nursery = ria_nursery` block deleted;
+  `_cancel_after_result_on_exit` comment refreshed.
+- `start_actor()`: `nursery=` param removed; body uses
+  `self._da_nursery.start(...)` directly.
+- `_open_and_supervise_one_cancels_all_nursery()`: the inner
+  `async with (collapse_eg(), trio.open_nursery() as
+  ria_nursery)` layer removed; `an = ActorNursery(actor,
+  da_nursery, errors)` constructed once under the single
+  `da_nursery`; the inner-try body de-indented one level;
+  both error handlers retained; the da-nursery lead comment
+  and the outer-`except` TODO refreshed to describe the
+  single-nursery reality + flag the (deferred) handler-merge.
+
+> `git diff 5cd190c5..wkt/to_actor_subpkg -- ai/conc-anal/ria_nursery_removal_plan.md`
+
+Added a "Step-B outcome" section (collapse rationale,
+handler-merge deferral, safety argument, gate result).
+
+## Test runs (verbatim)
+
+```
+targeted gate (trio):
+tests/test_cancellation.py tests/test_spawning.py
+tests/test_local.py tests/test_rpc.py tests/test_to_actor.py
+-> 49 passed, 1 xfailed in 88.62s
+
+signature checks:
+ActorNursery.__init__ params: ['self', 'actor', 'da_nursery', 'errors']
+start_actor has nursery param: False
+
+full-suite gate: in progress (see stepB_full.log)
+```

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

@@ -0,0 +1,68 @@
+---
+model: claude-opus-4-8
+service: claude
+session: c7860302-6a33-4d92-821c-0807d68be464
+timestamp: 2026-07-02T22:25:44Z
+git_ref: 9201a2ed
+scope: code
+substantive: true
+raw_file: 20260702T222544Z_9201a2ed_prompt_io.raw.md
+---
+
+## Prompt
+
+Continuing on branch `drop_ria_nursery` after step B
+(`9201a2ed`). User green-lit the deferred error-handler merge:
+
+> yes, i checked us out on a new branch for the ria-tn removal;
+> work off the new wkt branch drop_ria_nursery.
+
+and (re the migration) "nope, later we'll do this", plus:
+
+> do ensure the test suite is green in the current state and
+> those going forward ya?
+
+and the standing reminder: no commits without a human ack.
+
+## Response summary
+
+Step B2 of the `._ria_nursery` removal (issue #477): merge the
+two nested error handlers in
+`_open_and_supervise_one_cancels_all_nursery` into one, now
+that the 2ndary nursery (whose `__aexit__` the outer handler
+originally guarded) is gone.
+
+Trace-driven justification (see
+`ai/conc-anal/ria_nursery_removal_plan.md` Step-B2 section):
+the old inner handler records `errors[uid]` as its FIRST
+action, so `errors` is always non-empty by the time control
+could reach the outer handler — making the outer handler's
+`raise` DEAD (always superseded by the `finally`'s
+raise-from-`errors`). Collapsing to a single
+`except BaseException as _scope_err` + the existing `finally`
+is therefore behavior-preserving.
+
+- removed the outer `try` + the
+  `except (Exception, BaseExceptionGroup, trio.Cancelled)`
+  handler; de-indented the inner handler one level.
+- dropped the now-unused `outer_err`/`inner_err` locals
+  (single `_scope_err`).
+- net ~30 lines lighter.
+
+Gated hard: baseline (pre-B2 tip) = 300 passed + 1
+load-induced `TooSlowError` flake (passes 6/6 in isolation);
+B2 error/cancel gate (`test_cancellation remote_exc_relay
+inter_peer_cancellation advanced_faults oob_cancellation
+to_actor spawning local rpc`) = 71 passed, 1 xfailed; B2
+full-suite run following.
+
+## Files changed
+
+- `tractor/runtime/_supervise.py` — collapse the two handlers
+  into one; drop `outer_err`/`inner_err`
+
+## Human edits
+
+None yet — committed via the drafted
+`.claude/git_commit_msg_ria_b2.md` (user-driven
+`git commit --edit`).

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

@@ -0,0 +1,55 @@
+---
+model: claude-opus-4-8
+service: claude
+timestamp: 2026-07-02T22:25:44Z
+git_ref: 9201a2ed
+diff_cmd: git diff 9201a2ed..drop_ria_nursery
+---
+
+# Raw AI output (diff-ref mode)
+
+Step-B2 code lives on `drop_ria_nursery` after `9201a2ed`; per
+diff-ref mode the verbatim content is reachable via the pointer
+below.
+
+## Generated files
+
+> `git diff 9201a2ed..drop_ria_nursery -- tractor/runtime/_supervise.py`
+
+`_open_and_supervise_one_cancels_all_nursery`:
+- removed the outer `try:` wrapper and the
+  `except (Exception, BaseExceptionGroup, trio.Cancelled) as
+  _outer_err:` safety-net handler.
+- the former inner `except BaseException` is now THE handler,
+  renamed local `_inner_err` -> `_scope_err`, de-indented one
+  level; it sets `an._scope_error`, records `errors[uid]`,
+  waits on the debugger, `_join_procs.set()`, then a shielded
+  classify/log + snapshot-ria + `an.cancel()` + 0.5s-bounded
+  `_reap_ria_portals()`. No re-raise (the `finally` raises
+  from `errors`).
+- `finally` block unchanged.
+- dropped the `outer_err`/`inner_err` local decls at fn top.
+
+(The diff is large — ~119+/149- — because de-indenting the
+handler body one level rewrites every line in the block; the
+logic delta is just "two handlers -> one".)
+
+## Test runs (verbatim)
+
+```
+baseline (pre-B2, step-B tip 9201a2ed), full suite
+(dynamic_pub_sub deselected):
+1 failed, 300 passed, 9 skipped, 2 deselected, 1 xfailed,
+2 xpassed in 1499.49s
+  -> the 1 failure = test_ext_types_over_ipc[...] trio.TooSlowError
+     (load-induced; passes 6/6 in isolation in 4.89s)
+
+B2 error/cancel gate:
+tests/test_cancellation test_remote_exc_relay
+test_inter_peer_cancellation test_advanced_faults
+test_oob_cancellation test_to_actor test_spawning test_local
+test_rpc
+-> 71 passed, 1 xfailed in 125.26s
+
+B2 full-suite run: see b2_full.log
+```

ai/prompt-io/claude/20260706T172818Z_ad42871e_prompt_io.md

@@ -0,0 +1,81 @@
+---
+model: claude-fable-5
+service: claude
+session: 6db64ac6-6986-4505-9343-df4ee31e67db
+timestamp: 2026-07-06T17:28:18Z
+git_ref: ad42871e
+scope: code
+substantive: true
+raw_file: 20260706T172818Z_ad42871e_prompt_io.raw.md
+---
+
+## Prompt
+
+Resuming the `drop_ria_nursery` (wkt) branch after the human's
+review pause at `ba4af0ad`, with the directive:
+
+> we need to resolve the remaining outstandings ... to finish
+> this endeavour we ideally,
+> - finish the ria-tn hoist into the machinery in `to_actor` if
+>   possible.
+> - avoid supporting the `.run_in_actor()` non-blocking API
+>   going forward other then if we can re-wrap `to_actor.run()`
+>   per prior discussion with opus4.8.
+> - get the test harness completely green locally and in CI -
+>   AND ESPECIALLY without any hangs due to the "the old
+>   spawn-backend reaper raced `soft_kill()`'s scope-cancel"
+> - update all docs to reflect these changes once we're happy.
+
+plus mid-flight additions:
+
+> make sure both all the test and example code uses the naming
+> `an: ActorNursery` for variable names
+
+> where we at in terms of a potential hackaround design for
+> supporting a to_actor.run_async() ... without requiring the
+> same malarky ... with a `trio.Event` relay ... i'm totes fine
+> with rejigging the iface to NOT return a Portal as well.
+
+## Response summary
+
+Completed the entire #477 `run_in_actor` removal on
+`drop_ria_nursery` (16 commits this session), gated per-commit:
+
+- ported the remaining `test_cancellation` groups
+  (`some_cancels_all`, `nested_multierrors`+`spawn_and_error`,
+  SIGINT + sync-sleep) — one-shots become concurrent
+  `to_actor.run()` tasks in local task-nurseries; BEG-shape
+  assertions loosened for cancel-on-first + the runtime's
+  `collapse_eg()` single-member unwrap (a fully-raced nested
+  tree relays a bare annotated `RemoteActorError` chain).
+- fixed a pre-existing `UnboundLocalError` (`timeout` `match`
+  had no default arm for non-trio/MTF backends).
+- ported `test_dynamic_pub_sub`, 4 non-debugging examples, all
+  8 `debugging/` examples (debugger suite byte-identical,
+  28p/6s; `multi_subactors` introduces the "collect don't
+  cancel" reap-all replacement pattern), 8 docs pages + the
+  `experimental/_pubsub` docstring.
+- EXCISED the API + cluster: `run_in_actor`,
+  `_reap_ria_portals`, `_cancel_after_result_on_exit`,
+  `Portal._submit_for_result/_expect_result_ctx/
+  wait_for_result/result`, `exhaust_portal`,
+  `cancel_on_completion`, `NoResult` — net -402 lines. The
+  reap-hang class dissolves structurally (result-waits now only
+  in caller task-scope).
+- found + fixed a real migration race: mutual-rendezvous peers
+  (`test_trynamic_trio`, `a_trynamic_first_scene.py`) flaked
+  because an eagerly-reaped one-shot dies while its peer still
+  dials the registry-resolved (dead) sockaddr — such peers now
+  pin lifetimes via `start_actor()` + concurrent `Portal.run()`
+  + explicit `an.cancel()`.
+- `an: ActorNursery` naming sweep across tests/examples (±82
+  lines, scoped renames, prose untouched).
+- parked a `to_actor.open_one_shot()` design sketch (acm +
+  private task-nursery over blocking `run()`; done-Event as
+  memo not cancel-relay; no Portal) in the plan doc.
+
+## Files changed
+
+See commits `d01a2123..ad42871e` on `drop_ria_nursery`
+(tests, examples, docs, `tractor/{runtime,spawn,to_actor,msg}`
++ `_exceptions/_context/experimental`).

ai/prompt-io/claude/20260706T172818Z_ad42871e_prompt_io.raw.md

@@ -0,0 +1,39 @@
+---
+model: claude-fable-5
+service: claude
+timestamp: 2026-07-06T17:28:18Z
+git_ref: ad42871e
+diff_cmd: git diff ba4af0ad..ad42871e
+---
+
+# Raw AI output (diff-ref mode)
+
+This session's output spans the 16 migration/excision commits
+`d01a2123..ad42871e` on `drop_ria_nursery`; per diff-ref mode
+the verbatim content is reachable via the pointer below.
+
+## Generated files
+
+> `git diff ba4af0ad..ad42871e`
+
+Commit-wise (each `Gate:`-footed msg documents its own module
+gate):
+
+- `d01a2123` port `test_some_cancels_all`
+- `697c6152` fix unbound `timeout` (non-trio/MTF `match` arm)
+- `fa8799d5` port `test_nested_multierrors`
+- `f11754ce` port SIGINT + sync-sleep cancel tests
+- `cb6202e3` port `test_dynamic_pub_sub`
+- `d8af5f12` port non-debugging examples
+- `a3057cb2` port debugging examples (+ `test_debugger`
+  nested-nurseries final-shape expectations)
+- `d6bed7c4` port docs (8 rst pages)
+- `07e1669e` fix stale `@pub` docstring example
+- `2a59cefb` REMOVE `run_in_actor()` + the ria reap cluster
+  (net -402 lines)
+- `a297a32a` fix mutual-rendezvous premature-reap race
+- `ad42871e` `an: ActorNursery` naming sweep
+
+Plan/design record updated in
+`ai/conc-anal/ria_nursery_removal_plan.md` (RESOLVED section +
+the `to_actor.open_one_shot()` follow-up sketch).

ai/prompt-io/opencode/20260819T184640Z_481ba003_prompt_io.md

@@ -0,0 +1,56 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: 76c5d31c-5a2f-4503-9b16-410ee7f4fab3
+timestamp: 2026-08-19T18:46:40Z
+git_ref: 481ba003
+scope: code
+substantive: true
+raw_file: 20260819T184640Z_481ba003_prompt_io.raw.md
+---
+
+## Prompt
+
+Rebase PR #484 onto final PR #481, migrate every affected one-shot call
+to the new positional target API and continue through downstream tests,
+examples and documentation review.
+
+## Response summary
+
+Converted stale target keyword calls to target partials so previously
+named inputs remain explicit while placement/runtime controls stay
+direct. Updated error expectations for local signature validation and
+linked remote error propagation, then corrected docs which still
+described the removed one-shot implementation. Linked spawning and
+context lifecycle prose to the corresponding API methods and detailed
+context guide.
+
+## Files changed
+
+- `docs/api/core.rst` - describe linked one-shot context execution.
+- `docs/guide/rpc.rst` - update placement and target call semantics.
+- `docs/guide/spawning.rst` - document positional target inputs.
+- `examples/debugging/multi_nested_subactors_error_up_through_nurseries.py` - migrate nested actor target inputs.
+- `examples/debugging/root_cancelled_but_child_is_in_tty_lock.py` - preserve named recursive target inputs with partials.
+- `tests/test_advanced_streaming.py` - migrate streaming target inputs.
+- `tests/test_cancellation.py` - migrate calls and tighten errors.
+- `tests/test_infected_asyncio.py` - bind asyncio target options.
+- `tests/test_rpc.py` - migrate RPC target argument binding.
+- `tests/test_runtime.py` - preserve named runtime target inputs.
+- `tests/test_spawning.py` - preserve named spawning target inputs.
+
+## Human edits
+
+The human selected the stack order and final PR #481 base, asked the
+agent to continue after each diagnostic step and required a complete
+commit plan after independently force-pushing the rebased history.
+After reviewing the migration, the human required every formerly named
+target input to remain visibly named through `functools.partial()`
+rather than becoming positional. These were human-directed agent edits;
+the human also required plain `start_actor()` and `open_context()`
+references in the spawning and RPC guides to link to their API methods
+and the detailed context guide, then clarified that `to_actor.run()`
+already uses the full context API while `Portal.run()` should share
+linked lifecycle machinery without necessarily delegating through
+`Portal.open_context()` or adding a `Started` message. The human made
+no direct source-line edits.

ai/prompt-io/opencode/20260819T184640Z_481ba003_prompt_io.raw.md

@@ -0,0 +1,30 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-19T18:46:40Z
+git_ref: 481ba003
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Migrate PR #484's downstream one-shot calls to PR #481's final
+`tractor.to_actor.run()` contract after the stack rebase.
+
+> `git diff HEAD~1..HEAD -- docs examples tests`
+
+Pass target arguments positionally and bind target keyword-only inputs
+with `functools.partial()`. Keep placement and runtime controls as
+direct `to_actor.run()` keywords. Update the invalid-target-argument
+test to expect local signature binding before actor startup and require
+direct `RemoteActorError` propagation from linked one-shots.
+
+Update API and guide prose to describe positional target inputs,
+linked `Portal.open_context()` execution and per-child reaping instead
+of the removed `Portal.run()` and target-`**kwargs` conventions.
+
+Verification:
+
+- core and migrated runtime batches: `97 passed`
+- discovery and related lifecycle batch: `33 passed, 1 skipped`
+- changed executable examples: `9 passed`
+- mapped debugger cases: `12 passed, 6 skipped`
+- Ruff, compilation and `git diff --check`: clean

docs/api/core.rst

@@ -62,8 +62,8 @@ One-shot task actors
    ``portal=`` it owns only the linked task and leaves the existing
    actor's lifetime to the portal owner; that actor must expose both
    the target module and ``tractor.to_actor.MODULE``. It supersedes
-   the legacy, non-blocking ``ActorNursery.run_in_actor()`` retained
-   only for compatibility until its removal in PR #484.
+   the removed (legacy, non-blocking)
+   ``ActorNursery.run_in_actor()``.
 
 .. deprecated:: 0.1.0a6
 

examples/actor_spawning_and_causality_with_daemon.py

@@ -12,9 +12,9 @@ async def movie_theatre_question():
 async def main():
     """The main ``tractor`` routine.
     """
-    async with tractor.open_nursery() as n:
+    async with tractor.open_nursery() as an:
 
-        portal = await n.start_actor(
+        portal = await an.start_actor(
             'frank',
             # enable the actor to run funcs from this current module
             enable_modules=[__name__],

examples/asynchronous_generators.py

@@ -15,9 +15,9 @@ async def stream_forever() -> AsyncIterator[int]:
 
 async def main():
 
-    async with tractor.open_nursery() as n:
+    async with tractor.open_nursery() as an:
 
-        portal = await n.start_actor(
+        portal = await an.start_actor(
             'donny',
             enable_modules=[__name__],
         )

examples/debugging/open_ctx_modnofound.py

@@ -21,8 +21,8 @@ async def main() -> None:
 
     async with tractor.open_nursery(
         debug_mode=True,
-    ) as n:
-        portal = await n.start_actor(
+    ) as an:
+        portal = await an.start_actor(
             'ctx_child',
 
             # XXX: we don't enable the current module in order

examples/debugging/per_actor_debug.py

@@ -6,14 +6,14 @@ async def die():
 
 
 async def main():
-    async with tractor.open_nursery() as tn:
+    async with tractor.open_nursery() as an:
 
-        debug_actor = await tn.start_actor(
+        debug_actor = await an.start_actor(
             'debugged_boi',
             enable_modules=[__name__],
             debug_mode=True,
         )
-        crash_boi = await tn.start_actor(
+        crash_boi = await an.start_actor(
             'crash_boi',
             enable_modules=[__name__],
             # debug_mode=True,

examples/debugging/subactor_bp_in_ctx.py

@@ -58,8 +58,8 @@ async def main():
         debug_mode=True,
         enable_transports=[tpt],
         loglevel='devx',
-    ) as n:
-        p = await n.start_actor(
+    ) as an:
+        p = await an.start_actor(
             'bp_boi',
             enable_modules=[__name__],
         )

examples/infected_asyncio_echo_server.py

@@ -50,8 +50,8 @@ async def trio_to_aio_echo_server(
 
 async def main():
 
-    async with tractor.open_nursery() as n:
-        p = await n.start_actor(
+    async with tractor.open_nursery() as an:
+        p = await an.start_actor(
             'aio_server',
             enable_modules=[__name__],
             infect_asyncio=True,

examples/integration/open_context_and_sleep.py

@@ -29,9 +29,9 @@ async def main() -> None:
     ))
     await proc.wait()
     # await trio.sleep_forever()
-    # async with tractor.open_nursery() as n:
+    # async with tractor.open_nursery() as an:
 
-    #     portal = await n.start_actor(
+    #     portal = await an.start_actor(
     #         'rpc_server',
     #         enable_modules=[__name__],
     #     )

examples/parallelism/concurrent_actors_primes.py

@@ -55,7 +55,7 @@ async def worker_pool(workers=4):
     Yes, the workers stay alive (and ready for work) until you close
     the context.
     """
-    async with tractor.open_nursery() as tn:
+    async with tractor.open_nursery() as an:
 
         portals = []
         snd_chan, recv_chan = trio.open_memory_channel(len(PRIMES))
@@ -65,7 +65,7 @@ async def worker_pool(workers=4):
             # this starts a new sub-actor (process + trio runtime) and
             # stores it's "portal" for later use to "submit jobs" (ugh).
             portals.append(
-                await tn.start_actor(
+                await an.start_actor(
                     f'worker_{i}',
                     enable_modules=[__name__],
                 )
@@ -80,10 +80,10 @@ async def worker_pool(workers=4):
             async def send_result(func, value, portal):
                 await snd_chan.send((value, await portal.run(func, n=value)))
 
-            async with trio.open_nursery() as n:
+            async with trio.open_nursery() as tn:
 
                 for value, portal in zip(sequence, itertools.cycle(portals)):
-                    n.start_soon(
+                    tn.start_soon(
                         send_result,
                         worker_func,
                         value,
@@ -98,7 +98,7 @@ async def worker_pool(workers=4):
         yield _map
 
         # tear down all "workers" on pool close
-        await tn.cancel()
+        await an.cancel()
 
 
 async def main():

examples/rpc_bidir_streaming.py

@@ -31,9 +31,9 @@ async def simple_rpc(
 
 async def main() -> None:
 
-    async with tractor.open_nursery() as n:
+    async with tractor.open_nursery() as an:
 
-        portal = await n.start_actor(
+        portal = await an.start_actor(
             'rpc_server',
             enable_modules=[__name__],
         )

nooz/484.feature.rst

@@ -0,0 +1,5 @@
+Remove legacy ``ActorNursery.run_in_actor()``,
+``Portal.wait_for_result()`` and ``Portal.result()``. Use
+``tractor.to_actor.run()`` for caller-owned one-shot tasks,
+``Portal.run()`` for daemon RPC results or ``Portal.open_context()``
+for linked task dialogs.

tests/discovery/test_registrar.py

@@ -46,9 +46,9 @@ async def test_reg_then_unreg(
 
     async with tractor.open_nursery(
         registry_addrs=[reg_addr],
-    ) as n:
+    ) as an:
 
-        portal = await n.start_actor('actor', enable_modules=[__name__])
+        portal = await an.start_actor('actor', enable_modules=[__name__])
         uid = portal.channel.aid.uid
 
         async with tractor.get_registry(reg_addr) as aportal:
@@ -62,7 +62,7 @@ async def test_reg_then_unreg(
                 # XXX: can we figure out what the listen addr will be?
                 assert sockaddrs
 
-        await n.cancel()  # tear down nursery
+        await an.cancel()  # tear down nursery
 
         await trio.sleep(0.1)
         assert uid not in aportal.actor._registry
@@ -89,9 +89,9 @@ async def test_reg_then_unreg_maddr(
 
     async with tractor.open_nursery(
         registry_addrs=[maddr_str],
-    ) as n:
+    ) as an:
 
-        portal = await n.start_actor(
+        portal = await an.start_actor(
             'actor_maddr',
             enable_modules=[__name__],
         )
@@ -105,7 +105,7 @@ async def test_reg_then_unreg_maddr(
                 sockaddrs = actor._registry[uid]
                 assert sockaddrs
 
-        await n.cancel()
+        await an.cancel()
 
         await trio.sleep(0.1)
         assert uid not in aportal.actor._registry
@@ -152,23 +152,37 @@ async def test_trynamic_trio(
     for the directed subs.
 
     '''
-    async with tractor.open_nursery() as n:
+    async with tractor.open_nursery() as an:
         print("Alright... Action!")
 
-        donny = await n.run_in_actor(
-            ria_fn,
-            other_actor='gretchen',
-            reg_addr=reg_addr,
-            name='donny',
+        # donny + gretchen each wait on (then dial!) the *other*, so
+        # both actors must OUTLIVE both hellos: spawn as daemons and
+        # only reap after both tasks complete. NB a pair of eagerly
+        # reaped `to_actor.run()` one-shots races: the first to
+        # finish dies while the other may still be dialing its
+        # registry-resolved (now dead) sockaddr -> conn-refused.
+        portals: dict[str, tractor.Portal] = {
+            name: await an.start_actor(
+                name,
+                enable_modules=[__name__],
             )
-        gretchen = await n.run_in_actor(
+            for name in ('donny', 'gretchen')
+        }
+
+        async def _direct(this_name: str, other_actor: str):
+            res = await portals[this_name].run(
                 ria_fn,
-            other_actor='donny',
+                other_actor=other_actor,
                 reg_addr=reg_addr,
-            name='gretchen',
             )
-        print(await gretchen.result())
-        print(await donny.result())
+            print(res)
+
+        async with trio.open_nursery() as tn:
+            tn.start_soon(_direct, 'donny', 'gretchen')
+            tn.start_soon(_direct, 'gretchen', 'donny')
+
+        # both hellos have completed; reap the thespians.
+        await an.cancel()
         print("CUTTTT CUUTT CUT!!?! Donny!! You're supposed to say...")
 
 
@@ -270,13 +284,15 @@ async def spawn_and_check_registry(
                         portals = {}
                         for i in range(3):
                             name = f'a{i}'
-                            if with_streaming:
+                            # a daemon subactor is alive + registered
+                            # without a "main" task; the streaming
+                            # branch below uses the module funcs, the
+                            # non-streaming case just needs it up (was
+                            # `run_in_actor(trio.sleep_forever)`).
                             portals[name] = await an.start_actor(
-                                    name=name, enable_modules=[__name__])
-
-                            else:  # no streaming
-                                portals[name] = await an.run_in_actor(
-                                    trio.sleep_forever, name=name)
+                                name=name,
+                                enable_modules=[__name__],
+                            )
 
                         # wait on last actor to come up
                         async with tractor.wait_for_actor(name):

tests/test_advanced_streaming.py

@@ -3,6 +3,7 @@ Advanced streaming patterns using bidirectional streams and contexts.
 
 '''
 from collections import Counter
+from functools import partial
 import itertools
 import platform
 from typing import Type
@@ -24,6 +25,7 @@ _registry: dict[str, set[tractor.MsgStream]] = {
     'even': set(),
     'odd': set(),
 }
+_publisher_started: bool = False
 
 
 async def publisher(
@@ -32,11 +34,13 @@ async def publisher(
 
 ) -> None:
 
-    global _registry
+    global _publisher_started, _registry
 
     def is_even(i):
         return i % 2 == 0
 
+    _publisher_started = True
+    try:
         for val in itertools.count(seed):
 
             sub = 'even' if is_even(val) else 'odd'
@@ -48,6 +52,26 @@ async def publisher(
             # making it readable to a human user
             await trio.sleep(1/1000)
 
+    finally:
+        _publisher_started = False
+
+
+async def pubsub_active(
+    expected_subs: int,
+) -> bool:
+    '''
+    Report whether the publisher and all subscriber tasks are active.
+
+    Runs as an RPC task in the publisher actor, where `_registry` is
+    mutated by each `subscribe()` context after its consumer sends the
+    first subscription.
+
+    '''
+    return (
+        _publisher_started
+        and sum(map(len, _registry.values())) >= expected_subs
+    )
+
 
 @tractor.context
 async def subscribe(
@@ -173,8 +197,8 @@ def test_dynamic_pub_sub(
     # test. Picked backend-aware: under `trio` backend spawn is
     # cheap (~1s for `cpus` actors) but fork-based backends pay
     # a per-spawn cost (forkserver round-trip + IPC peer-handshake)
-    # that can stack up over `cpus - 1` sequential `n.run_in_actor()`
-    # calls — especially on UDS under cross-pytest contention
+    # that can stack up over the `cpus - 1` one-shot
+    # (`to_actor.run()`) spawns — especially on UDS under cross-pytest contention
     # (#451 / #452). 4s was flaking right at the edge under fork
     # backends — bumped to 8s with diag-snapshot-on-timeout via
     # `fail_after_w_trace` so a borderline run still fails loud
@@ -214,37 +238,71 @@ def test_dynamic_pub_sub(
                     f'enter `fail_after_w_trace({fail_after_s})` scope'
                 )
                 try:
-                    async with tractor.open_nursery(
+                    async with (
+                        tractor.open_nursery(
                             registry_addrs=[reg_addr],
                             debug_mode=debug_mode,
-                    ) as n:
+                        ) as an,
+                        # bg-schedules the forever-streaming
+                        # one-shots below; the user-cancel raise
+                        # cancels them all, each reaping its
+                        # subactor via `to_actor.run()`'s
+                        # (shielded) `Portal.cancel_actor()`.
+                        trio.open_nursery() as tn,
+                    ):
                         test_log.cancel(
                             'test_dynamic_pub_sub: '
                             'actor nursery opened'
                         )
 
                         # name of this actor will be same as target func
-                        await n.run_in_actor(publisher)
+                        tn.start_soon(
+                            partial(
+                                tractor.to_actor.run,
+                                publisher,
+                                an=an,
+                            )
+                        )
 
                         for i, sub in zip(
                             range(cpus - 2),
                             itertools.cycle(_registry.keys())
                         ):
-                            await n.run_in_actor(
+                            tn.start_soon(
+                                partial(
+                                    tractor.to_actor.run,
+                                    partial(
                                         consumer,
-                                name=f'consumer_{sub}',
                                         subs=[sub],
+                                    ),
+                                    an=an,
+                                    name=f'consumer_{sub}',
+                                )
                             )
 
                         # make one dynamic subscriber
-                        await n.run_in_actor(
+                        tn.start_soon(
+                            partial(
+                                tractor.to_actor.run,
+                                partial(
                                     consumer,
-                            name='consumer_dynamic',
                                     subs=list(_registry.keys()),
+                                ),
+                                an=an,
+                                name='consumer_dynamic',
+                            )
                         )
 
-                        # block until "cancelled by user"
-                        await trio.sleep(3)
+                        expected_subs: int = max(cpus - 2, 0) + 1
+                        async with tractor.wait_for_actor(
+                            'publisher',
+                        ) as portal:
+                            while not await portal.run(
+                                pubsub_active,
+                                expected_subs=expected_subs,
+                            ):
+                                await trio.sleep(0.01)
+
                         test_log.warning(
                             f'Raising user cancel exc: '
                             f'{expect_cancel_exc!r}'
@@ -347,10 +405,10 @@ def test_reqresp_ontopof_streaming():
             timeout = 4
 
         with trio.move_on_after(timeout):
-            async with tractor.open_nursery() as n:
+            async with tractor.open_nursery() as an:
 
                 # name of this actor will be same as target func
-                portal = await n.start_actor(
+                portal = await an.start_actor(
                     'dual_tasks',
                     enable_modules=[__name__]
                 )
@@ -413,9 +471,9 @@ def test_sigint_both_stream_types():
 
     async def main():
         with trio.fail_after(timeout):
-            async with tractor.open_nursery() as n:
+            async with tractor.open_nursery() as an:
                 # name of this actor will be same as target func
-                portal = await n.start_actor(
+                portal = await an.start_actor(
                     '2_way',
                     enable_modules=[__name__]
                 )
@@ -528,8 +586,8 @@ def test_local_task_fanout_from_stream(
 
         async with tractor.open_nursery(
             debug_mode=debug_mode,
-        ) as tn:
-            p: tractor.Portal = await tn.start_actor(
+        ) as an:
+            p: tractor.Portal = await an.start_actor(
                 'inf_streamer',
                 enable_modules=[__name__],
             )

tests/test_cancellation.py

@@ -2,6 +2,7 @@
 Cancellation and error propagation
 
 """
+from functools import partial
 import os
 import signal
 import platform
@@ -16,6 +17,10 @@ from tractor._testing import (
     tractor_test,
 )
 from tractor._testing.trace import FailAfterWTraceFactory
+from tractor.trionics import (
+    collapse_eg,
+    gather_contexts,
+)
 from .conftest import no_windows
 
 
@@ -68,10 +73,39 @@ async def assert_err(delay=0):
     assert 0
 
 
+@tractor.context
+async def assert_err_ctx(
+    ctx: tractor.Context,
+    delay: float = 0,
+) -> None:
+    '''
+    `@context` shim around `assert_err()` so the multi-actor error
+    tests can fan-out one-shot erroring subactors via
+    `Portal.open_context()` + `gather_contexts()` instead of the
+    removed `ActorNursery.run_in_actor()` (#477).
+
+    '''
+    await ctx.started()
+    await trio.sleep(delay)
+    assert 0
+
+
 async def sleep_forever():
     await trio.sleep_forever()
 
 
+@tractor.context
+async def sleep_forever_ctx(
+    ctx: tractor.Context,
+) -> None:
+    '''
+    Signal task startup before sleeping until context cancellation.
+
+    '''
+    await ctx.started()
+    await sleep_forever()
+
+
 async def do_nuthin():
     # just nick the scheduler
     await trio.sleep(0)
@@ -82,7 +116,7 @@ async def do_nuthin():
     [
         # expected to be thrown in assert_err
         ({}, AssertionError),
-        # argument mismatch raised in _invoke()
+        # argument mismatch rejected locally before spawn
         ({'unexpected': 10}, TypeError)
     ],
     ids=['no_args', 'unexpected_args'],
@@ -104,57 +138,38 @@ def test_remote_error(
     async def main():
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
-        ) as nursery:
+        ) as an:
 
-            # on a remote type error caused by bad input args
-            # this should raise directly which means we **don't** get
-            # an exception group outside the nursery since the error
-            # here and the far end task error are one in the same?
-            portal = await nursery.run_in_actor(
-                assert_err,
-                name='errorer',
-                **args
-            )
-
-            # get result(s) from main task
+            # `to_actor.run()` blocks on the one-shot's result and
+            # raises the remote error directly here in the caller's
+            # task. Invalid target args fail local signature binding
+            # before any one-shot actor is spawned.
             try:
-                # this means the root actor will also raise a local
-                # parent task error and thus an eg will propagate out
-                # of this actor nursery.
-                await portal.result()
+                await tractor.to_actor.run(
+                    partial(assert_err, **args),
+                    an=an,
+                    name='errorer',
+                )
             except tractor.RemoteActorError as err:
                 assert err.boxed_type == errtype
                 print("Look Maa that actor failed hard, hehh")
                 raise
 
-    # ensure boxed errors
+    # Invalid args never cross the process boundary.
     if args:
-        with pytest.raises(tractor.RemoteActorError) as excinfo:
+        with pytest.raises(errtype):
+            trio.run(main)
+
+    else:
+        # The linked one-shot raises the child's boxed error
+        # directly in this caller task.
+        with pytest.raises(
+            tractor.RemoteActorError,
+        ) as excinfo:
             trio.run(main)
 
         assert excinfo.value.boxed_type == errtype
 
-    else:
-        # the root task will also error on the `Portal.result()`
-        # call so we expect an error from there AND the child.
-        # |_ tho seems like on new `trio` this doesn't always
-        #    happen?
-        with pytest.raises((
-            BaseExceptionGroup,
-            tractor.RemoteActorError,
-        )) as excinfo:
-            trio.run(main)
-
-        # ensure boxed errors are `errtype`
-        err: BaseException = excinfo.value
-        if isinstance(err, BaseExceptionGroup):
-            suberrs: list[BaseException] = err.exceptions
-        else:
-            suberrs: list[BaseException] = [err]
-
-        for exc in suberrs:
-            assert exc.boxed_type == errtype
-
 
 def test_multierror(
     reg_addr: tuple[str, int],
@@ -162,113 +177,202 @@ def test_multierror(
     set_fork_aware_capture, #: Callable,
 ):
     '''
-    Verify we raise a ``BaseExceptionGroup`` out of a nursery where
-    more then one actor errors.
+    Verify concurrent one-shot subactors erroring propagate a remote
+    error out of the `gather_contexts()` fan-out — grouped as a
+    `BaseExceptionGroup`, or (under cancel-on-first, where the 2nd
+    errorer is cancelled before relaying its own exc) collapsed to a
+    single `RemoteActorError`.
+
+    NB the legacy `run_in_actor()` reaped *all* children at nursery
+    teardown so this always yielded a BEG-of-N; the `to_actor`
+    fan-out is cancel-on-first, so accept either shape.
 
     '''
     async def main():
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
-        ) as nursery:
+        ) as an:
 
-            await nursery.run_in_actor(assert_err, name='errorer1')
-            portal2 = await nursery.run_in_actor(assert_err, name='errorer2')
+            portals = [
+                await an.start_actor(
+                    f'errorer{i}',
+                    enable_modules=[__name__],
+                )
+                for i in range(2)
+            ]
 
-            # get result(s) from main task
-            try:
-                await portal2.result()
-            except tractor.RemoteActorError as err:
-                assert err.boxed_type is AssertionError
-                print("Look Maa that first actor failed hard, hehh")
-                raise
+            # both one-shot subactors error concurrently, so the
+            # `gather_contexts()` task-nursery collects them into a
+            # `BaseExceptionGroup` (was two non-blocking
+            # `run_in_actor()`s reaped at nursery teardown).
+            async with gather_contexts(
+                mngrs=[
+                    p.open_context(assert_err_ctx)
+                    for p in portals
+                ],
+            ):
+                pass
 
-        # here we should get a ``BaseExceptionGroup`` containing exceptions
-        # from both subactors
-
-    with pytest.raises(BaseExceptionGroup):
+    with pytest.raises((
+        BaseExceptionGroup,
+        tractor.RemoteActorError,
+    )) as excinfo:
         trio.run(main)
 
+    exc = excinfo.value
+    if isinstance(exc, tractor.RemoteActorError):
+        assert exc.boxed_type is AssertionError
+        return
+
+    def iter_group_leaves(
+        group: BaseExceptionGroup,
+    ):
+        for subexc in group.exceptions:
+            if isinstance(subexc, BaseExceptionGroup):
+                yield from iter_group_leaves(subexc)
+            else:
+                yield subexc
+
+    assertion_errors: list[tractor.RemoteActorError] = []
+    cancellations: list[BaseException] = []
+    for leaf in iter_group_leaves(exc):
+        if isinstance(leaf, tractor.ContextCancelled):
+            cancellations.append(leaf)
+        elif isinstance(leaf, trio.Cancelled):
+            cancellations.append(leaf)
+        else:
+            assert isinstance(leaf, tractor.RemoteActorError)
+            assert leaf.boxed_type is AssertionError
+            assertion_errors.append(leaf)
+
+    assert len(assertion_errors) in (1, 2)
+    if not cancellations:
+        assert len(assertion_errors) == 2
+
 
 @pytest.mark.parametrize(
-    'delay',
-    (0, 0.5),
-    ids='delays={}'.format,
+    'errorer_count',
+    (1, 5, 25),
+    ids='errorers={}'.format,
 )
-@pytest.mark.parametrize(
-    'num_subactors',
-    range(25, 26),
-    ids= 'num_subs={}'.format,
-)
-def test_multierror_fast_nursery(
-    reg_addr: tuple,
+def test_concurrent_start_error_reaps_all(
+    reg_addr: tuple[str, int],
     start_method: str,
-    num_subactors: int,
-    delay: float,
+    errorer_count: int,
     set_fork_aware_capture,
     fail_after_w_trace: FailAfterWTraceFactory,
 ):
     '''
-    Verify we raise a ``BaseExceptionGroup`` out of a nursery where
-    more then one actor errors and also with a delay before failure
-    to test failure during an ongoing spawning.
+    Reap high-fan-out children cancelled during concurrent startup.
+
+    The removed `test_multierror_fast_nursery()` launched 25 legacy
+    one-shots while earlier children could already be failing. Its
+    exact 25-error group depended on deferred nursery-exit result
+    collection, but the startup/cancellation load remains valuable.
+
+    This replacement schedules 25 blocking `to_actor.run()` calls and
+    holds all local caller tasks before any actor process starts.
+    Releasing one barrier makes the flat pool race through
+    `to_actor.run()` together; that API implicitly calls
+    `ActorNursery.start_actor()` for each child. Backend and Trio
+    scheduling then vary which children are spawning, handshaking or
+    running when the first remote error arrives.
+
+    Parameterizing one, five and all 25 errorers covers sparse through
+    saturated failure. Unlike the old nursery-owned deferred result
+    collection, the local Trio task nursery is the OCA supervisor: its
+    first remote error cancels sibling callers, and each
+    `to_actor.run()` shield-reaps any child process it accepted.
+    Bounded completion, only boxed assertion relays and empty
+    child/reap maps prove the entire flat pool remained supervised.
 
     '''
-    async def main():
-        # budget = 2× natural trio-backend cascade time for
-        # 25 errorer subactors (~14s observed). on-timeout
-        # diag snapshot → if the cancel cascade hangs
-        # (observed under MTF backend with N>=14 errorer
-        # subactors) we get a fresh ptree/wchan/py-spy dump
-        # on disk INSTEAD of an opaque pytest timeout-kill.
-        # See `tractor/_testing/trace.py` for the helper.
-        async with fail_after_w_trace(30.0):
+    child_count: int = 25
+
+    async def main() -> None:
+        callers_ready: int = 0
+        all_callers_ready = trio.Event()
+
+        async with fail_after_w_trace(40):
+            with pytest.raises((
+                BaseExceptionGroup,
+                tractor.RemoteActorError,
+            )) as excinfo:
                 async with tractor.open_nursery(
                     registry_addrs=[reg_addr],
-            ) as nursery:
+                ) as an:
+                    async def run_child(
+                        i: int,
+                    ) -> None:
+                        nonlocal callers_ready
 
-                for i in range(num_subactors):
-                    await nursery.run_in_actor(
-                        assert_err,
-                        name=f'errorer{i}',
-                        delay=delay
+                        # Hold every local caller before ANY actor
+                        # process starts. Arrival publication through
+                        # `.set()` has no checkpoint; releasing the
+                        # barrier maximizes scheduler/backend variation
+                        # across the implicit `start_actor()` calls.
+                        callers_ready += 1
+                        if callers_ready == child_count:
+                            all_callers_ready.set()
+                        await all_callers_ready.wait()
+
+                        is_errorer: bool = (
+                            i >= child_count - errorer_count
+                        )
+                        fn = (
+                            assert_err
+                            if is_errorer
+                            else sleep_forever
+                        )
+                        name = (
+                            f'errorer_{i}'
+                            if is_errorer
+                            else f'waiter_{i}'
+                        )
+                        await tractor.to_actor.run(
+                            fn,
+                            an=an,
+                            name=name,
                         )
 
-    # with pytest.raises(trio.MultiError) as exc_info:
-    # NOTE, `trio.TooSlowError` from `fail_after_w_trace`
-    # bubbles UN-wrapped if `open_nursery.__aexit__` never
-    # gets re-entered; wrapped inside a `BaseExceptionGroup`
-    # if it did. Accept both shapes so the matcher itself
-    # doesn't lie about *what* failed.
-    with pytest.raises(
-        (BaseExceptionGroup, trio.TooSlowError),
-    ) as exc_info:
+                    async with trio.open_nursery() as tn:
+                        for i in range(child_count):
+                            tn.start_soon(
+                                run_child,
+                                i,
+                            )
+
+            assert callers_ready == child_count
+            assert all_callers_ready.is_set()
+            assert not an._children
+            assert not an._child_reap_requests
+            assert not an._child_reaped
+
+        def iter_leaves(
+            exc: BaseException,
+        ):
+            if isinstance(exc, BaseExceptionGroup):
+                for subexc in exc.exceptions:
+                    yield from iter_leaves(subexc)
+            else:
+                yield exc
+
+        assertion_errors: list[tractor.RemoteActorError] = []
+        for leaf in iter_leaves(excinfo.value):
+            if isinstance(leaf, (
+                trio.Cancelled,
+                tractor.ContextCancelled,
+            )):
+                continue
+
+            assert isinstance(leaf, tractor.RemoteActorError)
+            assert leaf.boxed_type is AssertionError
+            assertion_errors.append(leaf)
+
+        assert 1 <= len(assertion_errors) <= errorer_count
+
     trio.run(main)
 
-    if isinstance(exc_info.value, trio.TooSlowError):
-        pytest.fail(
-            f'cancel cascade hung past 12s '
-            f'(num_subactors={num_subactors}, delay={delay}); '
-            f'see stderr for `fail_after_w_trace` snapshot path'
-        )
-
-    assert exc_info.type == ExceptionGroup
-    err = exc_info.value
-    exceptions = err.exceptions
-
-    if len(exceptions) == 2:
-        # sometimes oddly now there's an embedded BrokenResourceError ?
-        for exc in exceptions:
-            excs = getattr(exc, 'exceptions', None)
-            if excs:
-                exceptions = excs
-                break
-
-    assert len(exceptions) == num_subactors
-
-    for exc in exceptions:
-        assert isinstance(exc, tractor.RemoteActorError)
-        assert exc.boxed_type is AssertionError
-
 
 async def do_nothing():
     pass
@@ -296,16 +400,16 @@ def test_cancel_single_subactor(
         '''
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
-        ) as nursery:
+        ) as an:
 
-            portal = await nursery.start_actor(
+            portal = await an.start_actor(
                 'nothin', enable_modules=[__name__],
             )
             assert (await portal.run(do_nothing)) is None
 
             if mechanism == 'nursery_cancel':
                 # would hang otherwise
-                await nursery.cancel()
+                await an.cancel()
             else:
                 raise mechanism
 
@@ -337,8 +441,8 @@ async def test_cancel_infinite_streamer(
         trio.fail_after(4),
         trio.move_on_after(1) as cancel_scope
     ):
-        async with tractor.open_nursery() as n:
-            portal = await n.start_actor(
+        async with tractor.open_nursery() as an:
+            portal = await an.start_actor(
                 'donny',
                 enable_modules=[__name__],
             )
@@ -351,36 +455,36 @@ async def test_cancel_infinite_streamer(
 
     # we support trio's cancellation system
     assert cancel_scope.cancelled_caught
-    assert n.cancel_called
+    assert an.cancel_called
 
 
 @pytest.mark.parametrize(
     'num_actors_and_errs',
     [
-        # daemon actors sit idle while single task actors error out
+        # daemon actors sit idle while one-shot task actors error out
         (1, tractor.RemoteActorError, AssertionError, (assert_err, {}), None),
         (2, BaseExceptionGroup, AssertionError, (assert_err, {}), None),
         (3, BaseExceptionGroup, AssertionError, (assert_err, {}), None),
 
-        # 1 daemon actor errors out while single task actors sleep forever
+        # 1 daemon actor errors out while one-shot task actors sleep forever
         (3, tractor.RemoteActorError, AssertionError, (sleep_forever, {}),
          (assert_err, {}, True)),
-        # daemon actors error out after brief delay while single task
+        # daemon actors error out after brief delay while one-shot task
         # actors complete quickly
         (3, tractor.RemoteActorError, AssertionError,
          (do_nuthin, {}), (assert_err, {'delay': 1}, True)),
-        # daemon complete quickly delay while single task
+        # daemon complete quickly delay while one-shot task
         # actors error after brief delay
         (3, BaseExceptionGroup, AssertionError,
          (assert_err, {'delay': 1}), (do_nuthin, {}, False)),
     ],
     ids=[
-        '1_run_in_actor_fails',
-        '2_run_in_actors_fail',
-        '3_run_in_actors_fail',
+        '1_one_shot_fails',
+        '2_one_shots_fail',
+        '3_one_shots_fail',
         '1_daemon_actors_fail',
-        '1_daemon_actors_fail_all_run_in_actors_dun_quick',
-        'no_daemon_actors_fail_all_run_in_actors_sleep_then_fail',
+        '1_daemon_actors_fail_all_one_shots_dun_quick',
+        'no_daemon_actors_fail_all_one_shots_sleep_then_fail',
     ],
 )
 @tractor_test(
@@ -399,12 +503,34 @@ async def test_some_cancels_all(
 
     This is the first and only supervisory strategy at the moment.
 
+    One-shot subactors run as concurrent `to_actor.run()` tasks
+    in a local task-nursery so their errors raise WHILE the
+    actor-nursery block is still open (vs the legacy
+    `run_in_actor()` teardown-reap); the first error cancels the
+    sibling one-shots (whose `trio.Cancelled`s the task-nursery
+    absorbs) so the group shape is 1..num_actors
+    `RemoteActorError`s depending on relay-vs-cancel timing —
+    with `collapse_eg()` unwrapping the deterministic
+    single-error cases to a bare `RemoteActorError`.
+
+    Supervision and error flow (`RAE` is `RemoteActorError`):
+
+        root actor task
+        `-- ActorNursery an (owns child processes)
+            +-- daemon actor_i
+            |   `-- Portal.run() -------- RAE --+
+            +-- one-shot actor_i                 |
+            |   `-- remote target -- RAE --> run() caller
+            `-- local trio.Nursery tn <----------+
+                `-- first RAE cancels sibling callers
+                    `-- escapes -> an cancels/reaps all children
+
     '''
     (
         num_actors,
         first_err,
         err_type,
-        ria_func,
+        one_shot_func,
         da_func,
     ) = num_actors_and_errs
     try:
@@ -418,29 +544,35 @@ async def test_some_cancels_all(
                     enable_modules=[__name__],
                 ))
 
-            func, kwargs = ria_func
-            riactor_portals = []
+            func, kwargs = one_shot_func
+            async with (
+                collapse_eg(),
+                trio.open_nursery() as tn,
+            ):
                 for i in range(num_actors):
-                # start actor(s) that will fail immediately
-                riactor_portals.append(
-                    await an.run_in_actor(
-                        func,
+                    # schedule one-shot task actor(s); errors
+                    # raise into this task-nursery scope.
+                    tn.start_soon(
+                        partial(
+                            tractor.to_actor.run,
+                            partial(func, **kwargs),
+                            an=an,
                             name=f'actor_{i}',
-                        **kwargs
                         )
                     )
 
                 if da_func:
                     func, kwargs, expect_error = da_func
                     for portal in dactor_portals:
-                    # if this function fails then we should error here
-                    # and the nursery should teardown all other actors
+                        # if this function fails then we should error
+                        # here and the nursery should teardown all
+                        # other actors
                         try:
                             await portal.run(func, **kwargs)
 
                         except tractor.RemoteActorError as err:
                             assert err.boxed_type == err_type
-                        # we only expect this first error to propogate
+                            # we only expect this first error to propagate
                             # (all other daemons are cancelled before they
                             # can be scheduled)
                             num_actors = 1
@@ -449,20 +581,26 @@ async def test_some_cancels_all(
                         else:
                             if expect_error:
                                 pytest.fail(
-                                "Deamon call should fail at checkpoint?")
+                                    "Daemon call should fail at checkpoint?")
 
-        # should error here with a ``RemoteActorError`` or ``MultiError``
+    # should error here with a `RemoteActorError` or a beg of them
 
-    except first_err as _err:
+    except (
+        BaseExceptionGroup,
+        tractor.RemoteActorError,
+    ) as _err:
         err = _err
         if isinstance(err, BaseExceptionGroup):
-            assert len(err.exceptions) == num_actors
+            # only the concurrent multi-error cases can group; the
+            # relay-vs-cancel race means anywhere from 1 (all
+            # siblings cancelled before relaying) up to all
+            # `num_actors` errors may populate the group.
+            assert first_err is BaseExceptionGroup
+            assert 1 <= len(err.exceptions) <= num_actors
             for exc in err.exceptions:
-                if isinstance(exc, tractor.RemoteActorError):
+                assert isinstance(exc, tractor.RemoteActorError)
                 assert exc.boxed_type == err_type
         else:
-                    assert isinstance(exc, trio.Cancelled)
-        elif isinstance(err, tractor.RemoteActorError):
             assert err.boxed_type == err_type
 
         assert an.cancel_called is True
@@ -475,19 +613,33 @@ async def spawn_and_error(
     breadth: int,
     depth: int,
 ) -> None:
+    '''
+    Recursively spawn a breadth-wide level of erroring one-shot
+    subactors as concurrent `to_actor.run()` tasks; the leaf level
+    errors ~simultaneously and each level's task-nursery groups
+    whatever `RemoteActorError`s relay before the first one's
+    cancel wins, boxing the (`ExceptionGroup`-shaped) group into
+    this actor's own relayed error.
+
+    '''
     name = tractor.current_actor().name
-    async with tractor.open_nursery() as nursery:
+    async with (
+        tractor.open_nursery() as an,
+        trio.open_nursery() as tn,
+    ):
         for i in range(breadth):
 
             if depth > 0:
 
                 args = (
+                    partial(
                         spawn_and_error,
+                        breadth=breadth,
+                        depth=depth - 1,
+                    ),
                 )
                 kwargs = {
                     'name': f'spawner_{i}_depth_{depth}',
-                    'breadth': breadth,
-                    'depth': depth - 1,
                 }
             else:
                 args = (
@@ -496,7 +648,14 @@ async def spawn_and_error(
                 kwargs = {
                     'name': f'{name}_errorer_{i}',
                 }
-            await nursery.run_in_actor(*args, **kwargs)
+            tn.start_soon(
+                partial(
+                    tractor.to_actor.run,
+                    *args,
+                    an=an,
+                    **kwargs,
+                )
+            )
 
 
 # NOTE: `main_thread_forkserver` capture-fd hang class is no
@@ -538,7 +697,11 @@ async def test_nested_multierrors(
     depth: int,
 ):
     '''
-    Test that failed actor sets are wrapped in `BaseExceptionGroup`s.
+    Test that a nested tree of concurrently failing one-shot
+    subactors tears down cleanly, relaying (whatever subset of)
+    the leaf `AssertionError`s (that win the per-level
+    relay-vs-cancel race) re-boxed/grouped at each actor
+    boundary.
 
     Parametrized over recursion `depth ∈ {1, 3}`:
 
@@ -588,6 +751,13 @@ async def test_nested_multierrors(
     # fork-spawn jitter + UDS-contention widens both `t1` and
     # `t2` further.
     #
+    # NB post-#477 (`to_actor.run()` fan-out in a local
+    # task-nursery) a race-tripped sibling's `Cancelled` is
+    # ABSORBED by the task-nursery instead of landing in the
+    # group — the raced case now shows as a *smaller* BEG, so
+    # this marker should consistently `xpass`; drop it once CI
+    # confirms.
+    #
     # With `strict=False` the clean-cascade cases (most
     # depth=1 runs, rare depth=3 runs) report as `xpassed`
     # while the race-tripped cases report as `xfailed` —
@@ -672,6 +842,14 @@ async def test_nested_multierrors(
             timeout = 16
         case ('main_thread_forkserver', 3):
             timeout = 30
+        # any other fork-based backend (`mp_spawn` et al) pays
+        # the same per-spawn round-trip costs as MTF so rides
+        # its budgets; without a default arm `timeout` is left
+        # unbound -> `UnboundLocalError` at the scaling below.
+        case (_, 1):
+            timeout = 16
+        case (_, 3):
+            timeout = 30
 
     # inflate the budget by the throttle headroom probed above so
     # a slow box doesn't masquerade as a deadline regression.
@@ -684,66 +862,83 @@ async def test_nested_multierrors(
 
     async with fail_after_w_trace(timeout):
         try:
-            async with tractor.open_nursery() as nursery:
+            async with (
+                tractor.open_nursery() as an,
+                trio.open_nursery() as tn,
+            ):
                 for i in range(subactor_breadth):
-                    await nursery.run_in_actor(
+                    tn.start_soon(
+                        partial(
+                            tractor.to_actor.run,
+                            partial(
                                 spawn_and_error,
-                        name=f'spawner_{i}',
                                 breadth=subactor_breadth,
                                 depth=depth,
+                            ),
+                            an=an,
+                            name=f'spawner_{i}',
                         )
-        except BaseExceptionGroup as err:
-            assert len(err.exceptions) == subactor_breadth
-            for subexc in err.exceptions:
-
-                # verify first level actor errors are wrapped as remote
-                if _friggin_windows:
-
+                    )
+        except (
+            BaseExceptionGroup,
+            tractor.RemoteActorError,
+        ) as err:
+            # group membership is bounded by the relay-vs-cancel
+            # race: the first spawner-tree's error cancels its
+            # siblings, whose own errors only group when relayed
+            # first; a fully-raced tree even collapses (via the
+            # runtime's own `collapse_eg()` unwrapping each level's
+            # single-member group) to a bare `RemoteActorError`
+            # re-boxing the leaf `AssertionError` at every actor
+            # boundary. The deterministic exact-breadth nested-BEG
+            # was the legacy `run_in_actor()` reap-all-at-teardown.
+            subexcs: list[BaseException] = (
+                err.exceptions
+                if isinstance(err, BaseExceptionGroup)
+                else [err]
+            )
+            assert 1 <= len(subexcs) <= subactor_breadth
+            for subexc in subexcs:
+                if (
+                    _friggin_windows
+                    and
+                    isinstance(subexc, trio.Cancelled)
+                ):
                     # windows is often too slow and cancellation seems
                     # to happen before an actor is spawned
-                    if isinstance(subexc, trio.Cancelled):
                     continue
 
-                    elif isinstance(subexc, tractor.RemoteActorError):
-                        # on windows it seems we can't exactly be sure wtf
-                        # will happen..
-                        assert subexc.boxed_type in (
-                            tractor.RemoteActorError,
-                            trio.Cancelled,
-                            BaseExceptionGroup,
-                        )
-
-                    elif isinstance(subexc, BaseExceptionGroup):
-                        for subsub in subexc.exceptions:
-
-                            if subsub in (tractor.RemoteActorError,):
-                                subsub = subsub.boxed_type
-
-                            assert type(subsub) in (
-                                trio.Cancelled,
-                                BaseExceptionGroup,
-                            )
-                else:
                 assert isinstance(subexc, tractor.RemoteActorError)
 
-                if depth > 0 and subactor_breadth > 1:
-                    # XXX not sure what's up with this..
-                    # on windows sometimes spawning is just too slow and
-                    # we get back the (sent) cancel signal instead
-                    if _friggin_windows:
-                        if isinstance(subexc, tractor.RemoteActorError):
-                            assert subexc.boxed_type in (
-                                BaseExceptionGroup,
-                                tractor.RemoteActorError
-                            )
-                        else:
-                            assert isinstance(subexc, BaseExceptionGroup)
-                    else:
-                        assert subexc.boxed_type is ExceptionGroup
-                else:
-                    assert subexc.boxed_type in (
+                accepted: tuple[Type[BaseException], ...] = (
+                    # ≥2 sub-tree errors relayed before the
+                    # cancel-cascade won → grouped per-level.
+                    ExceptionGroup,
+                    # every level collapsed down to its lone
+                    # relayed (leaf) error.
+                    AssertionError,
+                    # a mid-level spawner relays an
+                    # already-boxed (collapsed) leaf chain,
+                    # re-boxing the `RemoteActorError` itself.
                     tractor.RemoteActorError,
-                        trio.Cancelled
+                    # under heavy load a runtime-internal reap
+                    # deadline can inject a `trio.Cancelled`
+                    # into a child's group before relay (the
+                    # same class the depth=3 throttle-xfail
+                    # covers) upgrading it from an
+                    # `ExceptionGroup`.
+                    BaseExceptionGroup,
+                )
+                if _friggin_windows:
+                    # on windows it seems we can't exactly be
+                    # sure wtf will happen..
+                    accepted += (
+                        trio.Cancelled,
+                    )
+                assert subexc.boxed_type in accepted
+        else:
+            pytest.fail(
+                'Should have raised a (grouped) `RemoteActorError`?'
             )
 
 
@@ -764,8 +959,8 @@ def test_cancel_via_SIGINT(
         with trio.fail_after(2):
             async with tractor.open_nursery(
                 registry_addrs=[reg_addr],
-            ) as tn:
-                await tn.start_actor('sucka')
+            ) as an:
+                await an.start_actor('sucka')
                 if 'mp' in start_method:
                     time.sleep(0.1)
                 os.kill(pid, signal.SIGINT)
@@ -809,12 +1004,24 @@ def test_cancel_via_SIGINT_other_task(
     ):
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
-        ) as tn:
-            for i in range(3):
-                await tn.run_in_actor(
-                    sleep_forever,
-                    name='namesucka',
+        ) as an:
+            portals = [
+                await an.start_actor(
+                    f'namesucka_{i}',
+                    enable_modules=[__name__],
                 )
+                for i in range(3)
+            ]
+
+            # Keep one linked RPC task active in every daemon before
+            # reporting startup, preserving the original
+            # `run_in_actor(sleep_forever)` cancellation target.
+            async with gather_contexts(
+                mngrs=[
+                    portal.open_context(sleep_forever_ctx)
+                    for portal in portals
+                ],
+            ):
                 task_status.started()
                 await trio.sleep_forever()
 
@@ -854,8 +1061,11 @@ async def spin_for(period=3):
 async def spawn_sub_with_sync_blocking_task():
     async with tractor.open_nursery() as an:
         print('starting sync blocking subactor..\n')
-        await an.run_in_actor(
+        # one-shot: parks HERE awaiting the sync-sleeping
+        # grandchild's result until cancelled from above.
+        await tractor.to_actor.run(
             spin_for,
+            an=an,
             name='sleeper',
         )
         print('exiting first subactor layer..\n')
@@ -961,11 +1171,19 @@ def test_cancel_while_childs_child_in_sync_sleep(
                     debug_mode=debug_mode,
                     registry_addrs=[reg_addr],
                 ) as an,
+                trio.open_nursery() as tn,
             ):
-                await an.run_in_actor(
+                # bg one-shot: parks on the middle actor's result
+                # (itself parked on the sync-sleeping grandchild)
+                # until the `assert 0` below cancels this scope.
+                tn.start_soon(
+                    partial(
+                        tractor.to_actor.run,
                         spawn_sub_with_sync_blocking_task,
+                        an=an,
                         name='sync_blocking_sub',
                     )
+                )
                 await trio.sleep(1)
 
                 if man_cancel_outer:
@@ -1013,8 +1231,8 @@ def test_fast_graceful_cancel_when_spawn_task_in_soft_proc_wait_for_daemon(
         start = time.time()
         try:
             async with trio.open_nursery() as nurse:
-                async with tractor.open_nursery() as tn:
-                    p = await tn.start_actor(
+                async with tractor.open_nursery() as an:
+                    p = await an.start_actor(
                         'fast_boi',
                         enable_modules=[__name__],
                     )

tests/test_child_manages_service_nursery.py

@@ -156,8 +156,8 @@ def test_actor_managed_trio_nursery_task_error_cancels_aio(
     async def main():
 
         # cancel the nursery shortly after boot
-        async with tractor.open_nursery() as n:
-            p = await n.start_actor(
+        async with tractor.open_nursery() as an:
+            p = await an.start_actor(
                 'nursery_mngr',
                 infect_asyncio=asyncio_mode,  # TODO, is this enabling debug mode?
                 enable_modules=[__name__],

tests/test_infected_asyncio.py

@@ -5,7 +5,7 @@ The hipster way to force SC onto the stdlib's "async": 'infection mode'.
 import asyncio
 import builtins
 from contextlib import ExitStack
-# from functools import partial
+from functools import partial
 import itertools
 import importlib
 import os
@@ -36,6 +36,7 @@ from tractor import (
     current_actor,
     Actor,
     to_asyncio,
+    to_actor,
     RemoteActorError,
     ContextCancelled,
 )
@@ -122,8 +123,9 @@ def test_trio_cancels_aio_on_actor_side(
                 registry_addrs=[reg_addr],
                 debug_mode=debug_mode,
             ) as an:
-                await an.run_in_actor(
+                await to_actor.run(
                     trio_cancels_single_aio_task,
+                    an=an,
                     infect_asyncio=True,
                 )
 
@@ -169,6 +171,28 @@ async def asyncio_actor(
         raise
 
 
+@tractor.context
+async def sleep_forever_aio_ctx(
+    ctx: tractor.Context,
+    expect_err: str = 'trio.Cancelled',
+) -> None:
+    '''
+    `@context` shim so a parent can spawn a forever-sleeping
+    infected-`asyncio` task via `Portal.open_context()` and cancel it
+    (via `Portal.cancel_actor()` or an enclosing `trio` cancel scope),
+    asserting the graceful `trio.Cancelled` teardown.
+
+    Replaces the legacy `ActorNursery.run_in_actor()` spawn the
+    aio-cancel tests below used to rely on (removed with #477).
+
+    '''
+    await ctx.started()
+    await asyncio_actor(
+        target='aio_sleep_forever',
+        expect_err=expect_err,
+    )
+
+
 def test_aio_simple_error(
     reg_addr: tuple[str, int],
     debug_mode: bool,
@@ -184,10 +208,13 @@ def test_aio_simple_error(
             registry_addrs=[reg_addr],
             debug_mode=debug_mode,
         ) as an:
-            await an.run_in_actor(
+            await to_actor.run(
+                partial(
                     asyncio_actor,
                     target='sleep_and_err',
                     expect_err='AssertionError',
+                ),
+                an=an,
                 infect_asyncio=True,
             )
 
@@ -219,18 +246,36 @@ def test_tractor_cancels_aio(
 
     '''
     async def main():
+        # anti-hang wall-clock cap: a per-test `trio.fail_after`
+        # is the blessed guard here since `pytest-timeout`'s
+        # global cap is intentionally off (see the `pyproject`
+        # NOTE — it breaks trio under fork backends). Generous +
+        # CPU-headroom-scaled bc this is an anti-hang guard, not
+        # a perf assertion; a wedged ria-reaper once hung this
+        # test forever (the `._ria_nursery`-removal regression).
+        # Regression fix: commit `d1fb4a1a`, documented in
+        # `ai/conc-anal/ria_nursery_removal_plan.md`.
+        from .conftest import cpu_perf_headroom
+        with trio.fail_after(9 * cpu_perf_headroom()):
             async with tractor.open_nursery(
                 debug_mode=debug_mode,
                 registry_addrs=[reg_addr],
             ) as an:
-            portal = await an.run_in_actor(
-                asyncio_actor,
-                target='aio_sleep_forever',
-                expect_err='trio.Cancelled',
+                p: tractor.Portal = await an.start_actor(
+                    'aio_daemon',
+                    enable_modules=[__name__],
                     infect_asyncio=True,
                 )
-            # cancel the entire remote runtime
-            await portal.cancel_actor()
+                async with (
+                    # `.cancel_actor()` below tears the ctx down
+                    expect_ctxc(yay=True),
+                    p.open_context(
+                        sleep_forever_aio_ctx,
+                    ) as (ctx, first),
+                ):
+                    # cancel the entire remote runtime while its
+                    # infected-`asyncio` task sleeps forever
+                    await p.cancel_actor()
 
     trio.run(main)
 
@@ -248,13 +293,19 @@ def test_trio_cancels_aio(
         with trio.move_on_after(1):
             async with tractor.open_nursery(
                 registry_addrs=[reg_addr],
-            ) as tn:
-                await tn.run_in_actor(
-                    asyncio_actor,
-                    target='aio_sleep_forever',
-                    expect_err='trio.Cancelled',
+            ) as an:
+                p: tractor.Portal = await an.start_actor(
+                    'aio_daemon',
+                    enable_modules=[__name__],
                     infect_asyncio=True,
                 )
+                async with p.open_context(
+                    sleep_forever_aio_ctx,
+                ) as (ctx, first):
+                    # block until the enclosing `move_on_after`
+                    # cancels this `trio` scope, tearing down the
+                    # infected-aio task via ctx cancellation
+                    await trio.sleep_forever()
 
     trio.run(main)
 
@@ -404,17 +455,20 @@ def test_aio_cancelled_from_aio_causes_trio_cancelled(
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
         ) as an:
-            p: tractor.Portal = await an.run_in_actor(
+            # `to_actor.run()` blocks on the one-shot's result and
+            # relays the remote error here in the caller's task.
+            with trio.fail_after(1 + delay):
+                await to_actor.run(
+                    partial(
                         asyncio_actor,
                         target='aio_cancel',
-                expect_err='tractor.to_asyncio.AsyncioCancelled',
+                        expect_err=(
+                            'tractor.to_asyncio.AsyncioCancelled'
+                        ),
+                    ),
+                    an=an,
                     infect_asyncio=True,
                 )
-            # NOTE: normally the `an.__aexit__()` waits on the
-            # portal's result but we do it explicitly here
-            # to avoid indent levels.
-            with trio.fail_after(1 + delay):
-                await p.wait_for_result()
 
     with pytest.raises(
         expected_exception=(RemoteActorError, ExceptionGroup),
@@ -615,13 +669,15 @@ def test_basic_interloop_channel_stream(
             async with tractor.open_nursery(
                 registry_addrs=[reg_addr],
             ) as an:
-                portal = await an.run_in_actor(
+                # should raise RAE directly
+                await to_actor.run(
+                    partial(
                         stream_from_aio,
-                    infect_asyncio=True,
                         fan_out=fan_out,
+                    ),
+                    an=an,
+                    infect_asyncio=True,
                 )
-                # should raise RAE diectly
-                await portal.result()
 
     trio.run(main)
 
@@ -634,13 +690,15 @@ def test_trio_error_cancels_intertask_chan(
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
         ) as an:
-            portal = await an.run_in_actor(
+            # should trigger remote actor error
+            await to_actor.run(
+                partial(
                     stream_from_aio,
                     trio_raise_err=True,
+                ),
+                an=an,
                 infect_asyncio=True,
             )
-            # should trigger remote actor error
-            await portal.result()
 
     with pytest.raises(RemoteActorError) as excinfo:
         trio.run(main)
@@ -670,14 +728,16 @@ def test_trio_closes_early_causes_aio_checkpoint_raise(
                 # enable_stack_on_sig=True,
                 registry_addrs=[reg_addr],
             ) as an:
-                portal = await an.run_in_actor(
+                # should raise RAE directly
+                print('waiting on final infected subactor result..')
+                res: None = await to_actor.run(
+                    partial(
                         stream_from_aio,
                         trio_exit_early=True,
+                    ),
+                    an=an,
                     infect_asyncio=True,
                 )
-                # should raise RAE diectly
-                print('waiting on final infected subactor result..')
-                res: None = await portal.wait_for_result()
                 assert res is None
                 print(f'infected subactor returned result: {res!r}\n')
 
@@ -721,15 +781,17 @@ def test_aio_exits_early_relays_AsyncioTaskExited(
                 debug_mode=debug_mode,
                 # enable_stack_on_sig=True,
             ) as an:
-                portal = await an.run_in_actor(
+                # should raise RAE directly
+                print('waiting on final infected subactor result..')
+                res: None = await to_actor.run(
+                    partial(
                         stream_from_aio,
-                    infect_asyncio=True,
                         trio_exit_early=False,
                         aio_exit_early=True,
+                    ),
+                    an=an,
+                    infect_asyncio=True,
                 )
-                # should raise RAE diectly
-                print('waiting on final infected subactor result..')
-                res: None = await portal.wait_for_result()
                 assert res is None
                 print(f'infected subactor returned result: {res!r}\n')
 
@@ -761,17 +823,21 @@ def test_aio_errors_and_channel_propagates_and_closes(
             registry_addrs=[reg_addr],
             debug_mode=debug_mode,
         ) as an:
-            portal = await an.run_in_actor(
+            # should trigger RAE directly, not an eg.
+            await to_actor.run(
+                partial(
                     stream_from_aio,
                     aio_raise_err=True,
+                ),
+                an=an,
                 infect_asyncio=True,
             )
-            # should trigger RAE directly, not an eg.
-            await portal.result()
 
     with pytest.raises(
-        # NOTE: bc we directly wait on `Portal.result()` instead
-        # of capturing it inside the `ActorNursery` machinery.
+        # NOTE: bc `to_actor.run()` blocks on + relays the result
+        # in the caller's task (not captured inside the
+        # `ActorNursery` teardown machinery) we get a direct RAE,
+        # not an eg.
         expected_exception=RemoteActorError,
     ) as excinfo:
         trio.run(main)

tests/test_inter_peer_cancellation.py

@@ -163,12 +163,12 @@ def test_do_not_swallow_error_before_started_by_remote_contextcancelled(
     async def main():
         async with tractor.open_nursery(
             debug_mode=debug_mode,
-        ) as n:
-            portal = await n.start_actor(
+        ) as an:
+            portal = await an.start_actor(
                 'errorer',
                 enable_modules=[__name__],
             )
-            await n.start_actor(
+            await an.start_actor(
                 'sleeper',
                 enable_modules=[__name__],
             )

tests/test_pubsub.py

@@ -139,9 +139,9 @@ async def test_required_args(callwith_expecterror):
         with pytest.raises(err):
             await func(**kwargs)
     else:
-        async with tractor.open_nursery() as n:
+        async with tractor.open_nursery() as an:
 
-            portal = await n.start_actor(
+            portal = await an.start_actor(
                 name='pubber',
                 enable_modules=[__name__],
             )
@@ -176,32 +176,73 @@ def test_multi_actor_subs_arbiter_pub(
 
     async def main():
 
-        async with tractor.open_nursery(
+        async with (
+            tractor.open_nursery(
                 registry_addrs=[reg_addr],
                 enable_modules=[__name__],
-        ) as n:
+            ) as an,
+            trio.open_nursery() as tn,
+        ):
 
             name = 'root'
 
             if pub_actor == 'streamer':
                 # start the publisher as a daemon
-                master_portal = await n.start_actor(
+                master_portal = await an.start_actor(
                     'streamer',
                     enable_modules=[__name__],
                 )
                 name = 'streamer'
 
-            even_portal = await n.run_in_actor(
+            root_uid = tractor.current_actor().aid.uid
+
+            # spawn the two subscriber actors as daemons and run
+            # `subs()` on each as a background task (was the legacy
+            # `run_in_actor()`); keep the portals for the explicit
+            # `cancel_actor()` teardown below. Each runner swallows
+            # only the cancellation relayed after its own teardown
+            # starts; every earlier or unrelated failure propagates.
+            async def _run_subs(
+                portal: tractor.Portal,
+                which: list[str],
+                teardown_started: trio.Event,
+            ) -> None:
+                try:
+                    await portal.run(
                         subs,
-                which=['even'],
-                name='evens',
-                pub_actor_name=name
+                        which=which,
+                        pub_actor_name=name,
                     )
-            odd_portal = await n.run_in_actor(
-                subs,
-                which=['odd'],
-                name='odds',
-                pub_actor_name=name
+                except tractor.ContextCancelled as ctxc:
+                    if not (
+                        teardown_started.is_set()
+                        and
+                        ctxc.canceller == root_uid
+                    ):
+                        raise
+
+            even_teardown_started = trio.Event()
+            odd_teardown_started = trio.Event()
+
+            even_portal = await an.start_actor(
+                'evens',
+                enable_modules=[__name__],
+            )
+            odd_portal = await an.start_actor(
+                'odds',
+                enable_modules=[__name__],
+            )
+            tn.start_soon(
+                _run_subs,
+                even_portal,
+                ['even'],
+                even_teardown_started,
+            )
+            tn.start_soon(
+                _run_subs,
+                odd_portal,
+                ['odd'],
+                odd_teardown_started,
             )
 
             async with tractor.wait_for_actor('evens'):
@@ -241,12 +282,14 @@ def test_multi_actor_subs_arbiter_pub(
             # await even_portal.result()
 
             await trio.sleep(0.5)
+            even_teardown_started.set()
             await even_portal.cancel_actor()
             await trio.sleep(1)
 
             if pub_actor == 'arbiter':
                 assert 'even' not in get_topics()
 
+            odd_teardown_started.set()
             await odd_portal.cancel_actor()
 
             if pub_actor == 'arbiter':
@@ -257,6 +300,9 @@ def test_multi_actor_subs_arbiter_pub(
             else:
                 await master_portal.cancel_actor()
 
+            # drop the bg `subs()` runners now the subs are cancelled
+            tn.cancel_scope.cancel()
+
     trio.run(main)
 
 
@@ -269,9 +315,9 @@ def test_single_subactor_pub_multitask_subs(
         async with tractor.open_nursery(
             registry_addrs=[reg_addr],
             enable_modules=[__name__],
-        ) as n:
+        ) as an:
 
-            portal = await n.start_actor(
+            portal = await an.start_actor(
                 'streamer',
                 enable_modules=[__name__],
             )

tests/test_ringbuf.py

@@ -143,9 +143,12 @@ def test_ringbuf(
                         child_read_shm,
                         **common_kwargs,
                         total_bytes=total_bytes,
-                    ) as (sctx, _sent),
+                    ) as (rctx, _sent),
                 ):
-                    await recv_p.result()
+                    # ctx-acm exits await each child task's
+                    # `Return` (the prior `recv_p.result()` here
+                    # was a daemon-portal no-op).
+                    pass
 
                 await send_p.cancel_actor()
                 await recv_p.cancel_actor()

tests/test_rpc.py

@@ -4,6 +4,7 @@ related API and error checks.
 
 '''
 import itertools
+from functools import partial
 from unittest.mock import (
     AsyncMock,
     Mock,
@@ -234,23 +235,26 @@ def test_rpc_errors(
             # do that if actually debugging subactor but keep it
             # disabled for the test.
             # debug_mode=True,
-        ) as n:
+        ) as an:
 
             actor = tractor.current_actor()
             assert actor.is_registrar
-            await n.run_in_actor(
+            await tractor.to_actor.run(
+                partial(
                     sleep_back_actor,
                     actor_name=subactor_requests_to,
+                    func_name=funcname,
+                    func_defined=bool(func_defined),
+                    exposed_mods=exposed_mods,
+                    reg_addr=reg_addr,
+                ),
+                an=an,
 
                 name='subactor',
 
-                # function from the local exposed module space
-                # the subactor will invoke when it RPCs back to this actor
-                func_name=funcname,
-                exposed_mods=exposed_mods,
-                func_defined=True if func_defined else False,
+                # Function from the local exposed module space the
+                # subactor invokes when it RPCs back to this actor.
                 enable_modules=subactor_exposed_mods,
-                reg_addr=reg_addr,
             )
 
     def run():

tests/test_runtime.py

@@ -2,6 +2,7 @@
 Verifying internal runtime state and undocumented extras.
 
 """
+from functools import partial
 import os
 
 import pytest
@@ -83,13 +84,14 @@ async def test_lifetime_stack_wipes_tmpfile(
             async with tractor.open_nursery(
                 loglevel=loglevel,
             ) as an:
-                await (  # inlined `tractor.Portal`
-                    await an.run_in_actor(
+                await tractor.to_actor.run(
+                    partial(
                         crash_and_clean_tmpdir,
                         tmp_file_path=path,
                         error=error_in_child,
+                    ),
+                    an=an,
                 )
-                ).result()
     except (
         tractor.RemoteActorError,
         BaseExceptionGroup,

tests/test_spawning.py

@@ -26,17 +26,31 @@ data_to_pass_down = {
 }
 
 
-async def spawn(
+async def run_same_func_in_child(
     should_be_root: bool,
     data: dict,
     reg_addr: tuple[str, int],
 
     debug_mode: bool = False,
 ):
+    '''
+    Invoke this same module-scoped RPC target in a child actor.
+
+    RPC targets cross IPC as `module:name` namespace paths, so this
+    helper must remain import-addressable at module scope instead of
+    being nested inside the test. The root invocation boots a runtime
+    and recursively calls this function as a one-shot child endpoint;
+    the child branch returns the result.
+
+    '''
     await trio.sleep(0.1)
     actor = tractor.current_actor(err_on_no_runtime=False)
 
-    if should_be_root:
+    if not should_be_root:
+        assert actor is not None
+        assert actor.is_registrar == should_be_root
+        return 10
+
     assert actor is None  # no runtime yet
     async with (
         tractor.open_root_actor(
@@ -48,44 +62,33 @@ async def spawn(
         actor: tractor.Actor = tractor.current_actor()
         assert actor.is_registrar == should_be_root
 
-            # spawns subproc here
-            portal: tractor.Portal = await an.run_in_actor(
-                fn=spawn,
+        # recursively spawn this same function as the lone
+        # task of a one-shot child subactor and get its result.
+        result = await tractor.to_actor.run(
+            partial(
+                run_same_func_in_child,
+                should_be_root=False,
+                data=data_to_pass_down,
+                reg_addr=reg_addr,
+            ),
+            an=an,
 
             # spawning args
             name='sub-actor',
             enable_modules=[__name__],
 
-                # passed to a subactor-recursive RPC invoke
-                # of this same `spawn()` fn.
-                should_be_root=False,
-                data=data_to_pass_down,
-                reg_addr=reg_addr,
         )
-
-            assert len(an._children) == 1
-            assert (
-                portal.channel.aid.uid
-                in
-                tractor.current_actor().ipc_server._peers
-            )
-
-            # get result from child subactor
-            result = await portal.result()
         assert result == 10
         return result
-    else:
-        assert actor.is_registrar == should_be_root
-        return 10
 
 
-def test_run_in_actor_same_func_in_child(
+def test_to_actor_run_same_func_in_child(
     reg_addr: tuple,
     debug_mode: bool,
 ):
     result = trio.run(
         partial(
-            spawn,
+            run_same_func_in_child,
             should_be_root=True,
             data=data_to_pass_down,
             reg_addr=reg_addr,
@@ -159,21 +162,18 @@ async def test_most_beautiful_word(
         async with tractor.open_nursery(
             debug_mode=debug_mode,
         ) as an:
-            portal = await an.run_in_actor(
+            res: Any = await tractor.to_actor.run(
+                partial(
                     cellar_door,
                     return_value=return_value,
+                ),
+                an=an,
                 name='some_linguist',
             )
-
-            res: Any = await portal.wait_for_result()
-            assert res == return_value
-    # The ``async with`` will unblock here since the 'some_linguist'
-    # actor has completed its main task ``cellar_door``.
-
-    # this should pull the cached final result already captured during
-    # the nursery block exit.
-    res: Any = await portal.wait_for_result()
             assert res == return_value
+    # The ``async with`` unblocks here — the 'some_linguist'
+    # one-shot actor completed its lone task ``cellar_door`` and
+    # was reaped by `to_actor.run()`.
     print(res)
 
 
@@ -215,11 +215,14 @@ def test_loglevel_propagated_to_subactor(
             start_method=start_method,
             registry_addrs=[reg_addr],
 
-        ) as tn:
-            await tn.run_in_actor(
+        ) as an:
+            await tractor.to_actor.run(
+                partial(
                     check_loglevel,
-                loglevel=level,
                     level=level,
+                ),
+                an=an,
+                loglevel=level,
             )
 
     trio.run(main)
@@ -267,11 +270,11 @@ async def check_parent_main_inheritance(
     return has_data
 
 
-def test_run_in_actor_can_skip_parent_main_inheritance(
+def test_to_actor_run_can_skip_parent_main_inheritance(
     start_method: str,  # <- only support on `trio` backend rn.
 ):
     '''
-    Verify ``inherit_parent_main=False`` on ``run_in_actor()``
+    Verify ``inherit_parent_main=False`` on ``to_actor.run()``
     prevents parent ``__main__`` data from reaching the child.
 
     '''
@@ -284,21 +287,25 @@ def test_run_in_actor_can_skip_parent_main_inheritance(
         async with tractor.open_nursery(start_method='trio') as an:
 
             # Default: child receives parent __main__ bootstrap data
-            replaying = await an.run_in_actor(
+            await tractor.to_actor.run(
+                partial(
                     check_parent_main_inheritance,
-                name='replaying-parent-main',
                     expect_inherited=True,
+                ),
+                an=an,
+                name='replaying-parent-main',
             )
-            await replaying.result()
 
             # Opt-out: child gets no parent __main__ data
-            isolated = await an.run_in_actor(
+            await tractor.to_actor.run(
+                partial(
                     check_parent_main_inheritance,
+                    expect_inherited=False,
+                ),
+                an=an,
                 name='isolated-parent-main',
                 inherit_parent_main=False,
-                expect_inherited=False,
             )
-            await isolated.result()
 
     trio.run(main)
 

tests/test_to_actor.py

@@ -1,8 +1,8 @@
 '''
 `tractor.to_actor`: one-shot single-remote-task API suite.
 
-Verifies the "spiritual successor" to (and eventual
-replacement of) `ActorNursery.run_in_actor()`; see
+Verifies the "spiritual successor" to (and replacement of)
+the removed legacy `ActorNursery.run_in_actor()`; see
 https://github.com/goodboy/tractor/issues/477
 
 '''
@@ -160,7 +160,7 @@ async def test_remote_error_relayed_to_caller_task(
     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.
+    removed legacy `.run_in_actor()` API.
 
     '''
     with pytest.raises(RemoteActorError) as excinfo:

tractor/_context.py

@@ -780,7 +780,7 @@ class Context:
         # `Portal.open_context()` has been opened since it's
         # assumed that other portal APIs like,
         #  - `Portal.run()`,
-        #  - `ActorNursery.run_in_actor()`
+        #  - `to_actor.run()`
         # do their own error checking at their own call points and
         # result processing.
 

tractor/_exceptions.py

@@ -1161,10 +1161,6 @@ class TransportClosed(Exception):
         )
 
 
-class NoResult(RuntimeError):
-    "No final result is expected for this actor"
-
-
 class ModuleNotExposed(ModuleNotFoundError):
     "The requested module is not exposed for RPC"
 

tractor/experimental/_pubsub.py

@@ -217,20 +217,23 @@ def pub(
 
     .. code:: python
 
-        from functools import partial
         import tractor
 
         async with tractor.open_nursery() as n:
-            portal = n.run_in_actor(
+            portal = await n.start_actor(
                 'publisher',  # actor name
-                partial(      # func to execute in it
-                    pub_service,
+                enable_modules=[__name__],
+            )
+            try:
+                async with portal.open_stream_from(
+                    pub_service,  # func to execute in it
                     topics=('clicks', 'users'),
                     task_name='source1',
-                )
-            )
-            async for value in await portal.result():
+                ) as stream:
+                    async for value in stream:
                         print(f"Subscriber received {value}")
+            finally:
+                await portal.cancel_actor()
 
 
     Here, you don't need to provide the ``ctx`` argument since the

tractor/msg/types.py

@@ -306,12 +306,11 @@ class Start(
 
     It is called by all the following public APIs:
 
-    - `ActorNursery.run_in_actor()`
+    - `to_actor.run()`
 
     - `Portal.run()`
           `|_.run_from_ns()`
           `|_.open_stream_from()`
-          `|_._submit_for_result()`
 
     - `Context.open_context()`
 

tractor/runtime/_portal.py

@@ -50,13 +50,11 @@ from ..ipc import Channel
 from ..log import get_logger
 from ..msg import (
     # Error,
-    PayloadMsg,
     NamespacePath,
     Return,
 )
 from .._exceptions import (
     ActorTooSlowError,
-    NoResult,
     TransportClosed,
 )
 from .._context import (
@@ -102,14 +100,6 @@ class Portal:
     ) -> None:
 
         self._chan: Channel = channel
-        # during the portal's lifetime
-        self._final_result_pld: Any|None = None
-        self._final_result_msg: PayloadMsg|None = None
-
-        # When set to a ``Context`` (when _submit_for_result is called)
-        # it is expected that ``result()`` will be awaited at some
-        # point.
-        self._expect_result_ctx: Context|None = None
         self._streams: set[MsgStream] = set()
 
         # TODO, this should be PRIVATE (and never used publicly)! since it's just
@@ -137,102 +127,6 @@ class Portal:
         )
         return self.chan
 
-    # TODO: factor this out into a `.highlevel` API-wrapper that uses
-    # a single `.open_context()` call underneath.
-    async def _submit_for_result(
-        self,
-        ns: str,
-        func: str,
-        **kwargs
-    ) -> None:
-
-        if self._expect_result_ctx is not None:
-            raise RuntimeError(
-                'A pending main result has already been submitted'
-            )
-
-        self._expect_result_ctx: Context = await self.actor.start_remote_task(
-            self.channel,
-            nsf=NamespacePath(f'{ns}:{func}'),
-            kwargs=kwargs,
-            portal=self,
-        )
-
-    # TODO: we should deprecate this API right? since if we remove
-    # `.run_in_actor()` (and instead move it to a `.highlevel`
-    # wrapper api (around a single `.open_context()` call) we don't
-    # really have any notion of a "main" remote task any more?
-    #
-    # @api_frame
-    async def wait_for_result(
-        self,
-        hide_tb: bool = True,
-    ) -> Any:
-        '''
-        Return the final result delivered by a `Return`-msg from the
-        remote peer actor's "main" task's `return` statement.
-
-        '''
-        __tracebackhide__: bool = hide_tb
-        # Check for non-rpc errors slapped on the
-        # channel for which we always raise
-        exc = self.channel._exc
-        if exc:
-            raise exc
-
-        # not expecting a "main" result
-        if self._expect_result_ctx is None:
-            peer_id: str = f'{self.channel.aid.reprol()!r}'
-            log.warning(
-                f'Portal to peer {peer_id} will not deliver a final result?\n'
-                f'\n'
-                f'Context.result() can only be called by the parent of '
-                f'a sub-actor when it was spawned with '
-                f'`ActorNursery.run_in_actor()`'
-                f'\n'
-                f'Further this `ActorNursery`-method-API will deprecated in the'
-                f'near fututre!\n'
-            )
-            return NoResult
-
-        # expecting a "main" result
-        assert self._expect_result_ctx
-
-        if self._final_result_msg is None:
-            try:
-                (
-                    self._final_result_msg,
-                    self._final_result_pld,
-                ) = await self._expect_result_ctx._pld_rx.recv_msg(
-                    ipc=self._expect_result_ctx,
-                    expect_msg=Return,
-                )
-            except BaseException as err:
-                # TODO: wrap this into `@api_frame` optionally with
-                # some kinda filtering mechanism like log levels?
-                __tracebackhide__: bool = False
-                raise err
-
-        return self._final_result_pld
-
-    # TODO: factor this out into a `.highlevel` API-wrapper that uses
-    # a single `.open_context()` call underneath.
-    async def result(
-        self,
-        *args,
-        **kwargs,
-    ) -> Any|Exception:
-        typname: str = type(self).__name__
-        log.warning(
-            f'`{typname}.result()` is DEPRECATED!\n'
-            f'\n'
-            f'Use `{typname}.wait_for_result()` instead!\n'
-        )
-        return await self.wait_for_result(
-            *args,
-            **kwargs,
-        )
-
     async def _cancel_streams(self):
         # terminate all locally running async generator
         # IPC calls

tractor/runtime/_supervise.py

@@ -20,7 +20,6 @@
 """
 from contextlib import asynccontextmanager as acm
 from functools import partial
-import inspect
 from typing import (
     TYPE_CHECKING,
 )
@@ -225,7 +224,6 @@ class ActorNursery:
         self,
         # TODO: maybe def these as fields of a struct looking type?
         actor: Actor,
-        ria_nursery: trio.Nursery,
         da_nursery: trio.Nursery,
         errors: dict[tuple[str, str], BaseException],
 
@@ -267,16 +265,6 @@ class ActorNursery:
         # and syncing purposes to any actor opened nurseries.
         self._implicit_runtime_started: bool = False
 
-        # TODO: remove the `.run_in_actor()` API and thus this 2ndary
-        # nursery when that API get's moved outside this primitive!
-        self._ria_nursery = ria_nursery
-
-        # TODO, factor this into a .hilevel api!
-        #
-        # portals spawned with ``run_in_actor()`` are
-        # cancelled when their "main" result arrives
-        self._cancel_after_result_on_exit: set = set()
-
         # trio.Nursery-like cancel (request) statuses
         self._cancelled_caught: bool = False
         self._cancel_called: bool = False
@@ -438,11 +426,6 @@ class ActorNursery:
         debug_mode: bool|None = None,
         infect_asyncio: bool = False,
         inherit_parent_main: bool = True,
-
-        # TODO: ideally we can rm this once we no longer have
-        # a `._ria_nursery` since the dependent APIs have been
-        # removed!
-        nursery: trio.Nursery|None = None,
         proc_kwargs: dict[str, typing.Any] | None = None,
 
     ) -> Portal:
@@ -510,10 +493,8 @@ class ActorNursery:
 
         # start a task to spawn a process
         # blocks until process has been started and a portal setup
-        nursery: trio.Nursery = nursery or self._da_nursery
-
         # XXX: the type ignore is actually due to a `mypy` bug
-        return await nursery.start(  # type: ignore
+        return await self._da_nursery.start(  # type: ignore
             partial(
                 _spawn.new_proc,
                 name,
@@ -528,86 +509,6 @@ class ActorNursery:
             )
         )
 
-    # TODO: DEPRECATE THIS:
-    # -[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,
-
-        fn: typing.Callable,
-        *,
-
-        name: str | None = None,
-        bind_addrs: UnwrappedAddress|None = None,
-        rpc_module_paths: list[str] | None = None,
-        enable_modules: list[str] | None = None,
-        loglevel: str | None = None,  # set log level per subactor
-        infect_asyncio: bool = False,
-        inherit_parent_main: bool = True,
-        proc_kwargs: dict[str, typing.Any] | None = None,
-
-        **kwargs,  # explicit args to ``fn``
-
-    ) -> Portal:
-        '''
-        Spawn a new actor, run a lone task, then terminate the actor and
-        return its result.
-
-        Actors spawned using this method are kept alive at nursery teardown
-        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__
-
-        if name is None:
-            # use the explicit function name if not provided
-            name = fn.__name__
-
-        proc_kwargs = dict(proc_kwargs or {})
-        portal: Portal = await self.start_actor(
-            name,
-            enable_modules=[mod_path] + (
-                enable_modules or rpc_module_paths or []
-            ),
-            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
-        )
-
-        # XXX: don't allow stream funcs
-        if not (
-            inspect.iscoroutinefunction(fn) and
-            not getattr(fn, '_tractor_stream_function', False)
-        ):
-            raise TypeError(f'{fn} must be an async function!')
-
-        # this marks the actor to be cancelled after its portal result
-        # is retreived, see logic in `open_nursery()` below.
-        self._cancel_after_result_on_exit.add(portal)
-        await portal._submit_for_result(
-            mod_path,
-            fn.__name__,
-            **kwargs
-        )
-        return portal
-
     # @api_frame
     async def cancel(
         self,
@@ -738,40 +639,20 @@ async def _open_and_supervise_one_cancels_all_nursery(
     # normally don't need to show user by default
     __tracebackhide__: bool = hide_tb
 
-    outer_err: BaseException|None = None
-    inner_err: BaseException|None = None
-
     # the collection of errors retreived from spawned sub-actors
     errors: dict[tuple[str, str], BaseException] = {}
 
-    # This is the outermost level "deamon actor" nursery. It is awaited
-    # **after** the below inner "run in actor nursery". This allows for
-    # handling errors that are generated by the inner nursery in
-    # a supervisor strategy **before** blocking indefinitely to wait for
-    # actors spawned in "daemon mode" (aka started using
-    # `ActorNursery.start_actor()`).
-
-    # errors from this daemon actor nursery bubble up to caller
+    # The single "daemon actor" nursery into which ALL subactors
+    # are spawned; one-shot (`to_actor.run()`) subactors are
+    # result-waited and reaped in their caller's own task-scope
+    # (see the #477 `.run_in_actor()`/`._ria_nursery` removal);
+    # errors from this nursery bubble up to the caller.
     async with (
         collapse_eg(),
         trio.open_nursery() as da_nursery,
-    ):
-        try:
-            # This is the inner level "run in actor" nursery. It is
-            # awaited first since actors spawned in this way (using
-            # `ActorNusery.run_in_actor()`) are expected to only
-            # return a single result and then complete (i.e. be canclled
-            # gracefully). Errors collected from these actors are
-            # immediately raised for handling by a supervisor strategy.
-            # As such if the strategy propagates any error(s) upwards
-            # the above "daemon actor" nursery will be notified.
-            async with (
-                collapse_eg(),
-                trio.open_nursery() as ria_nursery,
     ):
         an = ActorNursery(
             actor,
-                    ria_nursery,
             da_nursery,
             errors
         )
@@ -789,9 +670,19 @@ async def _open_and_supervise_one_cancels_all_nursery(
             )
             an._request_reap_all()
 
-                except BaseException as _inner_err:
-                    inner_err = _inner_err
-                    errors[actor.aid.uid] = inner_err
+        # Single one-cancels-all handler for the (now single)
+        # daemon nursery. Pre-#477 a 2ndary `._ria_nursery`
+        # required a separate *outer* handler to catch errors
+        # bubbling from its task-reaping `__aexit__`; with that
+        # nursery gone this lone handler covers every scope
+        # error. NB: we deliberately do NOT re-raise here — the
+        # `finally` below raises the collected `errors` (as a
+        # single exc or `BaseExceptionGroup`), which already
+        # superseded the old outer handler's `raise` anyway
+        # since `errors` is populated (below) before any await.
+        except BaseException as _scope_err:
+            an._scope_error = _scope_err
+            errors[actor.aid.uid] = _scope_err
 
             # If we error in the root but the debugger is
             # engaged we don't want to prematurely kill (and
@@ -814,12 +705,12 @@ async def _open_and_supervise_one_cancels_all_nursery(
             # block here might not complete?  For now,
             # shield both.
             with trio.CancelScope(shield=True):
-                        etype: type = type(inner_err)
+                etype: type = type(_scope_err)
                 if etype in (
                     trio.Cancelled,
                     KeyboardInterrupt,
                 ) or (
-                            is_multi_cancelled(inner_err)
+                    is_multi_cancelled(_scope_err)
                 ):
                     log.cancel(
                         f'Actor-nursery cancelled by {etype}\n\n'
@@ -836,7 +727,7 @@ async def _open_and_supervise_one_cancels_all_nursery(
                     log.cancel(
                         'Actor-nursery caught remote cancellation\n'
                         '\n'
-                                f'{inner_err.tb_str}'
+                        f'{_scope_err.tb_str}'
                     )
                 else:
                     log.exception(
@@ -855,50 +746,11 @@ async def _open_and_supervise_one_cancels_all_nursery(
                 # cancel all subactors
                 await an.cancel()
 
-            # ria_nursery scope end
-
-        # TODO: this is the handler around the ``.run_in_actor()``
-        # nursery. Ideally we can drop this entirely in the future as
-        # the whole ``.run_in_actor()`` API should be built "on top of"
-        # this lower level spawn-request-cancel "daemon actor" API where
-        # a local in-actor task nursery is used with one-to-one task
-        # + `await Portal.run()` calls and the results/errors are
-        # handled directly (inline) and errors by the local nursery.
-        except (
-            Exception,
-            BaseExceptionGroup,
-            trio.Cancelled
-        ) as _outer_err:
-            outer_err = _outer_err
-
-            an._scope_error = outer_err or inner_err
-
-            # XXX: yet another guard before allowing the cancel
-            # sequence in case a (single) child is in debug.
-            await debug.maybe_wait_for_debugger(
-                child_in_debug=an._at_least_one_child_in_debug
-            )
-
-            # If actor-local error was raised while waiting on
-            # ".run_in_actor()" actors then we also want to cancel all
-            # remaining sub-actors (due to our lone strategy:
-            # one-cancels-all).
-            if an._children:
-                log.cancel(
-                    'Actor-nursery cancelling due error type:\n'
-                    f'{outer_err}\n'
-                )
-                with trio.CancelScope(shield=True):
-                    await an.cancel()
-            raise
-
         finally:
-            # No errors were raised while awaiting ".run_in_actor()"
-            # actors but those actors may have returned remote errors as
-            # results (meaning they errored remotely and have relayed
-            # those errors back to this parent actor). The errors are
-            # collected in ``errors`` so cancel all actors, summarize
-            # all errors and re-raise.
+            # an error was stashed by the handler above (or by
+            # a spawn task via the shared `errors` dict) so
+            # cancel any remaining subactors, summarize and
+            # re-raise.
             if errors:
                 if an._children:
                     with trio.CancelScope(shield=True):
@@ -942,15 +794,13 @@ async def open_nursery(
     Create and yield a new ``ActorNursery`` to be used for spawning
     structured concurrent subactors.
 
-    When an actor is spawned a new trio task is started which
-    invokes one of the process spawning backends to create and start
-    a new subprocess. These tasks are started by one of two nurseries
-    detailed below. The reason for spawning processes from within
-    a new task is because ``trio_run_in_process`` itself creates a new
-    internal nursery and the same task that opens a nursery **must**
-    close it. It turns out this approach is probably more correct
-    anyway since it is more clear from the following nested nurseries
-    which cancellation scopes correspond to each spawned subactor set.
+    When an actor is spawned a new trio task invokes one of the
+    process spawning backends to create and start a new subprocess.
+    These tasks are started in the supervisor's process nursery.
+    Spawning from a task is required because ``trio_run_in_process``
+    creates an internal nursery which the opening task **must** close;
+    this also makes each task's cancellation scope correspond to its
+    spawned subactor.
 
     '''
     __tracebackhide__: bool = hide_tb

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,
 )
@@ -199,15 +198,6 @@ async def mp_proc(
         with trio.CancelScope(shield=True):
             await reap_request.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.
@@ -217,13 +207,6 @@ async def mp_proc(
             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/_spawn.py

@@ -126,98 +126,6 @@ def try_set_start_method(
     return _ctx
 
 
-async def exhaust_portal(
-
-    portal: Portal,
-    actor: Actor
-
-) -> Any:
-    '''
-    Pull final result from portal (assuming it has one).
-
-    If the main task is an async generator do our best to consume
-    what's left of it.
-    '''
-    __tracebackhide__ = True
-    try:
-        log.debug(
-            f'Waiting on final result from {actor.aid.uid}'
-        )
-
-        # XXX: streams should never be reaped here since they should
-        # always be established and shutdown using a context manager api
-        final: Any = await portal.wait_for_result()
-
-    except (
-        Exception,
-        BaseExceptionGroup,
-    ) as err:
-        # we reraise in the parent task via a ``BaseExceptionGroup``
-        return err
-
-    except trio.Cancelled as err:
-        # lol, of course we need this too ;P
-        # TODO: merge with above?
-        log.warning(
-            'Cancelled portal result waiter task:\n'
-            f'uid: {portal.channel.aid}\n'
-            f'error: {err}\n'
-        )
-        return err
-
-    else:
-        log.debug(
-            f'Returning final result from portal:\n'
-            f'uid: {portal.channel.aid}\n'
-            f'result: {final}\n'
-        )
-        return final
-
-
-async def cancel_on_completion(
-
-    portal: Portal,
-    actor: Actor,
-    errors: dict[tuple[str, str], Exception],
-
-) -> None:
-    '''
-    Cancel actor gracefully once its "main" portal's
-    result arrives.
-
-    Should only be called for actors spawned via the
-    `Portal.run_in_actor()` API.
-
-    => and really this API will be deprecated and should be
-    re-implemented as a `.hilevel.one_shot_task_nursery()`..)
-
-    '''
-    # if this call errors we store the exception for later
-    # in ``errors`` which will be reraised inside
-    # an exception group and we still send out a cancel request
-    result: Any|Exception = await exhaust_portal(
-        portal,
-        actor,
-    )
-    if isinstance(result, Exception):
-        errors[actor.aid.uid]: Exception = result
-        log.cancel(
-            'Cancelling subactor runtime due to error:\n\n'
-            f'Portal.cancel_actor() => {portal.channel.aid}\n\n'
-            f'error: {result}\n'
-        )
-
-    else:
-        log.runtime(
-            'Cancelling subactor gracefully:\n\n'
-            f'Portal.cancel_actor() => {portal.channel.aid}\n\n'
-            f'result: {result}\n'
-        )
-
-    # cancel the process now that we have a final result
-    await portal.cancel_actor()
-
-
 async def hard_kill(
     proc: trio.Process,
 
@@ -464,8 +372,8 @@ async def new_proc(
 
 
 # NOTE: bottom-of-module to avoid a circular import since the
-# backend submodules pull `cancel_on_completion`/`soft_kill`/
-# `hard_kill`/`proc_waiter` from this module.
+# backend submodules pull `soft_kill`/`hard_kill`/`proc_waiter`
+# from this module.
 from ._trio import trio_proc
 from ._mp import mp_proc
 

tractor/spawn/_trio.py

@@ -49,7 +49,6 @@ from tractor.msg import (
     pretty_struct,
 )
 from ._spawn import (
-    cancel_on_completion,
     hard_kill,
     soft_kill,
 )
@@ -215,15 +214,6 @@ async def trio_proc(
         with trio.CancelScope(shield=True):
             await reap_request.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.
@@ -233,14 +223,6 @@ async def trio_proc(
             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

@@ -25,8 +25,8 @@ its result and (when the call owns the subactor) reap it.
 Target arguments follow Trio's positional convention; use
 `functools.partial()` to bind target keyword arguments.
 
-The "spiritual successor" to (and eventual replacement of)
-the `ActorNursery.run_in_actor()` API; see
+The "spiritual successor" to (and replacement of) the removed
+legacy `ActorNursery.run_in_actor()` API; see
 https://github.com/goodboy/tractor/issues/477
 
 '''

tractor/to_actor/_api.py

@@ -31,7 +31,7 @@ the lower level daemon-actor spawn + portal APIs,
 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)
+spawn-machinery nurseries as with the (now removed) legacy
 `ActorNursery.run_in_actor()` API.
 
 '''
@@ -282,8 +282,8 @@ async def run(
     module in its `enable_modules` list. Calls that spawn their own
     actor add the trampoline module automatically.
 
-    Unlike `ActorNursery.run_in_actor()` (which returns
-    a `Portal` whose result is only collected at
+    Unlike the removed legacy `ActorNursery.run_in_actor()` (which
+    returned a `Portal` whose result was 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