Make sub-spawn strategy toggleable in `we_are_processes`

The #470 boot-latency example hard-coded spawning each `worker_<i>`
subactor concurrently from a bg `trio.Task` (so each child's cold
`import tractor` overlaps). Add a `main()` `spawn_subs_in_bg_tasks`
flag so the serial-spawn path can be demo'd/compared too: flip it
`False` to `start_actor()` each sub inline in the loop before
handing the ready `Portal` to the bg task.

Deats,
- factor an `open_ep(ptl, i)` helper out of `spawn_and_open_ep()` -
  just the `Portal.open_context()` + `wait_for_result()` half, now
  that the spawn step is caller-optional.
- `spawn_and_open_ep()` grows a `maybe_ptl: Portal|None = None`
  param: spawn the subactor itself when unset (bg-task path), OW
  reuse the pre-spawned one (serial path).
- move the "overlap cold imports" rationale comment onto the new
  `main()` param where the toggle now lives.

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

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

@@ -1,80 +0,0 @@
----
-model: claude-fable-5
-service: claude
-session: f4da6654-63f9-4641-b567-f200da46d697
-timestamp: 2026-07-02T16:16:24Z
-git_ref: 65bf9df5
-scope: tests
-substantive: true
-raw_file: 20260702T161624Z_65bf9df5_prompt_io.raw.md
----
-
-## Prompt
-
-> attempt to resolve
-> https://github.com/goodboy/tractor/issues/474
-> do it with /open-wkt.
-
-Issue #474: "We need test suite for
-`.trionics.start_or_cancel()`" — dig out the use case
-from `modden` as a unit test (and provide motivation
-for upstream `trio` to possibly provide better
-porcelain around this case). The wrapper landed via
-PR #464 with no tests.
-
-## Response summary
-
-New `tests/trionics/test_taskc.py` (9 tests) covering
-`start_or_cancel()`, authored in wkt
-`start_or_cancel_tests_474`.
-
-Behaviour was first probed empirically (trio 0.29):
-the lossy `RuntimeError('child exited without calling
-task_status.started()')` only fires when the child
-exits pre-`.started()` WITHOUT propagating the ambient
-`Cancelled` — i.e. when the child (or lib code it
-calls) absorbs the cancel in a graceful-teardown
-pattern; a well-behaved child surfaces `Cancelled`
-straight out of `.start()`. The `modden`
-`progman.open_wks()` use case was reconstructed from
-`modden/runtime/progman.py` accordingly.
-
-Tests (each `use_start_or_cancel` parametrization also
-pins upstream trio's current lossy behaviour as
-wart-documentation):
-
-- `test_sibling_err_not_masked_by_startup_rte` — the
-  `modden` case: sibling error OOB-cancels the shared
-  nursery scope; with the wrapper ONLY the root-cause
-  `ValueError` escapes; bare `.start()` adds the lossy
-  RTE alongside.
-- `test_pure_oob_cancel_not_morphed_to_rte` — plain
-  ancestor `cs.cancel()`: wrapper → clean exit; bare
-  → eg-wrapped RTE.
-- `test_genuine_startup_rte_still_raised` — no
-  cancellation → protocol-bug RTE re-raised same as
-  bare.
-- `test_childs_own_rte_never_demoted_to_cancel` — a
-  child's own `RuntimeError('never got started!')` /
-  `RuntimeError(1234)` under ambient cancel is never
-  demoted to `Cancelled` (exact-msg-match + str-guard
-  regression cover).
-- `test_started_value_and_args_passthru` — happy path:
-  positional args, `name=`, `.started()` value.
-
-Verified: 9/9 pass; 0 flakes across 50 hammer runs;
-two impl mutations (checkpoint removed; guard relaxed
-to substring match) each caught by exactly the
-targeted tests; `tests/trionics/` +
-`tests/test_trioisms.py` subset green (23 passed,
-5 xfailed); ruff clean; 69-col style.
-
-## Files changed
-
-- `tests/trionics/test_taskc.py` — new
-  `start_or_cancel()` unit-test suite (gh #474).
-
-## Human edits
-
-Pending review — session paused pre-commit per user
-deadline; nothing committed as of this entry.

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

@@ -1,107 +0,0 @@
----
-model: claude-fable-5
-service: claude
-timestamp: 2026-07-02T16:16:24Z
-git_ref: 65bf9df5
-diff_cmd: git diff main..wkt/start_or_cancel_tests_474
----
-
-# Raw output — gh #474 `start_or_cancel()` test suite
-
-## Generated test code
-
-> `git diff main..wkt/start_or_cancel_tests_474 -- tests/trionics/test_taskc.py`
-
-Prose summary of the generated module
-(`tests/trionics/test_taskc.py`):
-
-- module docstring framing the `trio.Nursery.start()`
-  startup-cancellation wart, the wrapper's repair, and
-  the intent that `use_start_or_cancel=False` params
-  double as upstream-trio wart-documentation (break on
-  a trio upgrade → upstream may have shipped porcelain,
-  re-audit the wrapper); cites gh #474 / PR #464 and
-  `modden`'s `progman.open_wks()` as the source use
-  case.
-- shared children: `absorbs_cancel_pre_started()` (the
-  graceful-teardown cancel-absorber which triggers the
-  lossy RTE path) + `raise_val_err()` (fast-erroring
-  sibling).
-- `test_sibling_err_not_masked_by_startup_rte`
-  (parametrized `use_start_or_cancel`): asserts eg
-  contains exactly one `ValueError` and, wrapper-case,
-  NO residual RTE (`eg.split(ValueError)` remainder is
-  `None`); bare-case, the residual RTE carries trio's
-  exact "child exited without calling" wording.
-- `test_pure_oob_cancel_not_morphed_to_rte`
-  (parametrized): wrapper-case runs clean and asserts
-  `cs.cancelled_caught`; bare-case asserts the
-  eg-wrapped RTE.
-- `test_genuine_startup_rte_still_raised`
-  (parametrized): no-cancel protocol bug → RTE with
-  trio's wording from both call forms.
-- `test_childs_own_rte_never_demoted_to_cancel`
-  (parametrized `rte_arg` in `'never got started!'`,
-  `1234`): child cancels the ambient scope then raises
-  its own RTE synchronously (no checkpoint between →
-  deterministically under-cancellation at catch time);
-  asserts the RTE survives with `args[0]` intact.
-- `test_started_value_and_args_passthru`: `.started()`
-  value, positional args and the `name=` kwarg (via
-  `trio.lowlevel.current_task().name`) all forward.
-
-## Non-code output (verbatim highlights)
-
-Behaviour probe (trio 0.29, scratchpad scripts) — the
-decision basis for the test shapes:
-
-```
-== B-sibling-err use_soc=False
-  start raised: RuntimeError('child exited without
-    calling task_status.started()')
-  top-level: ExceptionGroup([ValueError('sibling blew
-    up!'), RuntimeError('child exited without calling
-    task_status.started()')])
-== B-cs-cancel use_soc=False
-  top-level: ExceptionGroup([RuntimeError('child
-    exited without calling task_status.started()')])
-== B-sibling-err use_soc=True
-  start raised: Cancelled()
-  top-level: ExceptionGroup([ValueError('sibling blew
-    up!')])
-== B-cs-cancel use_soc=True
-  start raised: Cancelled()
-  top-level: clean return
-== own-rte-under-cancel (both) -> RTE('never got
-  started!') propagates unchanged
-```
-
-Key finding: with a WELL-BEHAVED (non-absorbing) child
-an OOB ancestor cancel surfaces `Cancelled` directly
-from `.start()` on trio 0.29 — the lossy RTE requires
-the child to absorb its cancel pre-`.started()`, which
-is what `modden`'s `open_from_wks` teardown did. Trio's
-nursery-exit wait defers cancel delivery to children,
-so all tested shapes are deterministic (0 flakes / 50
-runs).
-
-Mutation verification:
-
-```
-mutation 1 (checkpoint_if_cancelled removed):
-  FAILED test_sibling_err_not_masked_by_startup_rte[True]
-  FAILED test_pure_oob_cancel_not_morphed_to_rte[True]
-mutation 2 (guard relaxed to 'started' substring,
-            isinstance dropped):
-  FAILED test_childs_own_rte_never_demoted_to_cancel[never got started!]
-  FAILED test_childs_own_rte_never_demoted_to_cancel[1234]
-```
-
-Final runs:
-
-```
-tests/trionics/test_taskc.py: 9 passed in 0.03s
-hammer: 0/50 runs failed
-tests/trionics/ + tests/test_trioisms.py:
-  23 passed, 5 xfailed in 3.02s
-```

ai/prompt-io/opencode/20260804T030309Z_65bf9df5_prompt_io.md

@@ -1,39 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-session: moc-teardown-completion-20260804
-timestamp: 2026-08-04T03:03:09Z
-git_ref: 65bf9df5
-scope: code
-substantive: true
-raw_file: 20260804T030309Z_65bf9df5_prompt_io.raw.md
----
-
-## Prompt
-
-Patch Tractor's `maybe_open_context()` so the final consumer waits for
-resource `__aexit__()` completion and receives cleanup errors. Reuse
-`outcome.Outcome` for the exit result; use the smaller mutable
-`_CtxExit` holder if that makes the implementation simpler. Add and run
-the relevant existing unit tests, but do not commit or push the patch.
-After reviewing the result, simplify `_CtxExit` back to an optional
-exception because the success outcome carries no useful value.
-
-## Response summary
-
-Added an exception-backed completion handshake between
-`_Cache.run_ctx()` and the final `maybe_open_context()` consumer.
-Serialized consumer registration and final teardown under the per-key
-lock, preserving that lock for queued entrants. Added deterministic
-regressions for normal exit, cleanup errors, cancellation interactions,
-service-nursery cancellation, and teardown re-entry.
-
-## Files changed
-
-- `tractor/trionics/_mngrs.py` - publish and unwrap cached exit outcomes.
-- `tests/test_resource_cache.py` - cover completion and cancellation.
-
-## Human edits
-
-The user directed the final simplification from `outcome.Outcome` to an
-optional exception field. The patch remains uncommitted.

ai/prompt-io/opencode/20260804T030309Z_65bf9df5_prompt_io.raw.md

@@ -1,36 +0,0 @@
----
-model: openai/gpt-5.6-sol
-service: opencode
-timestamp: 2026-08-04T03:03:09Z
-git_ref: 65bf9df5
-diff_cmd: git diff HEAD~1..HEAD
----
-
-Implemented cached-context exit completion in
-`tractor.trionics.maybe_open_context()`.
-
-> `git diff HEAD~1..HEAD -- tractor/trionics/_mngrs.py`
-
-The generated implementation adds `_CtxExit`, whose `done` event
-publishes an `outcome.Outcome[None]`. `_Cache.run_ctx()` records either
-`Value(None)` or `Error(exc)` after the resource exit attempt. The final
-MOC consumer signals `no_more_users`, waits for completion under a
-shielded cancel scope, removes the per-key lock, and unwraps the outcome
-so ordinary cleanup failures are raised at the consumer boundary.
-`trio.Cancelled`, `KeyboardInterrupt`, and `SystemExit` continue through
-the service task rather than being converted into regular cleanup
-errors.
-
-> `git diff HEAD~1..HEAD -- tests/test_resource_cache.py`
-
-The generated regressions cover successful exit blocking, cleanup-error
-delivery, final-user cancellation, cancellation combined with a cleanup
-error, and service-nursery cancellation. The existing teardown re-entry
-test now uses explicit events instead of a ten-second cleanup sleep and
-asserts that the replacement resource is a fresh cache miss.
-
-Verification:
-
-`env PYTHONPATH="$PWD" /home/goodboy/repos/tractor/py313/bin/python -m pytest tests/test_resource_cache.py`
-
-Result: `16 passed in 8.37s`.

ai/prompt-io/opencode/README.md

@@ -1,22 +0,0 @@
-# AI Prompt I/O Log - OpenCode
-
-This directory tracks prompt inputs and model outputs for AI-assisted
-development using `opencode`.
-
-## Policy
-
-Prompt logging follows the [NLNet generative AI policy][nlnet-ai]. All
-substantive AI contributions are logged with:
-
-- Model name and version
-- Timestamps
-- The prompts that produced the output
-- Unedited model output (`.raw.md` files)
-
-[nlnet-ai]: https://nlnet.nl/foundation/policies/generativeAI/
-
-## Usage
-
-Entries are created by the prompt-io workflow. Human contributors remain
-accountable for all decisions. AI-generated content is never presented as
-human-authored work.

examples/parallelism/we_are_processes.py

@@ -23,18 +23,10 @@ async def endpoint(
     await trio.sleep_forever()
 
 
-async def spawn_and_open_ep(
-    an: tractor.ActorNursery,
+async def open_ep(
+    ptl: tractor.Portal,
     i: int,
 ) -> None:
-    '''
-    Spawn a subactor, start a remote `endpoint()`-task in it.
-
-    '''
-    ptl: tractor.Portal = await an.start_actor(
-        name=f'worker_{i}',
-        enable_modules=[__name__],
-    )
     ctx: tractor.Context
     async with ptl.open_context(endpoint) as (
         ctx,
@@ -47,7 +39,33 @@ async def spawn_and_open_ep(
         await ctx.wait_for_result()
 
 
-async def main():
+async def spawn_and_open_ep(
+    an: tractor.ActorNursery,
+    i: int,
+    maybe_ptl: tractor.Portal|None = None,
+) -> None:
+    '''
+    Spawn a subactor, start a remote `endpoint()`-task in it.
+
+    '''
+    if maybe_ptl is None:
+        maybe_ptl: tractor.Portal = await an.start_actor(
+            name=f'worker_{i}',
+            enable_modules=[__name__],
+        )
+    await open_ep(
+        ptl=maybe_ptl,
+        i=i,
+    )
+
+
+async def main(
+    # spawn subs concurrently (in bg `trio.Task`s) so each
+    # actor's cold `import tractor` (~0.4s, see #470) overlaps
+    # instead of stacking; once forkserver (#463) lands, spawn
+    # is cheap enough to just loop sequentially.
+    spawn_subs_in_bg_tasks: bool = True,
+):
     '''
     Spawn a subactor-per-CPU then self-destruct the cluster.
 
@@ -60,17 +78,21 @@ async def main():
             # https://github.com/goodboy/tractor/pull/463
             # start_method='main_thread_forkserver',
         ) as an,
-        # spawn subs concurrently (in bg `trio.Task`s) so each
-        # actor's cold `import tractor` (~0.4s, see #470) overlaps
-        # instead of stacking; once forkserver (#463) lands, spawn
-        # is cheap enough to just loop sequentially.
         trio.open_nursery() as tn,
     ):
         for i in range(cpu_count()):
+
+            maybe_ptl: tractor.Portal|None = None
+            if not spawn_subs_in_bg_tasks:
+                maybe_ptl: tractor.Portal = await an.start_actor(
+                    name=f'worker_{i}',
+                    enable_modules=[__name__],
+                )
             tn.start_soon(
                 spawn_and_open_ep,
                 an,
                 i,
+                maybe_ptl,
             )
         destruct_in: int = 2
         print(

tests/test_log_sys.py

@@ -2,19 +2,16 @@
 `tractor.log`-wrapping unit tests.
 
 '''
-import logging
 from pathlib import Path
 import shutil
 from types import ModuleType
 
 import pytest
 import tractor
-import trio
 from tractor import (
     _code_load,
     log,
 )
-from tractor.ipc import _chan
 
 
 def test_root_pkg_not_duplicated_in_logger_name():
@@ -225,88 +222,6 @@ def test_add_log_level_pluggable():
             delattr(log.StackLevelAdapter, name.lower())
 
 
-@pytest.mark.parametrize(
-    'suppression',
-    [
-        'level',
-        'logger',
-        'global',
-    ],
-)
-def test_log_guard_skips_payload_formatting(
-    monkeypatch: pytest.MonkeyPatch,
-    suppression: str,
-):
-    '''
-    Suppressed transport logs must not render payloads.
-
-    The original hot-path guard compared only the effective logger
-    level. A logger disabled through its `Logger.disabled` flag or
-    the global `logging.disable()` threshold could therefore still
-    call `pformat()` before `Logger.isEnabledFor()` discarded the
-    record.
-
-    Exercise effective-level, per-logger, and global suppression
-    independently. A poisoned `_chan.pformat()` proves rendering is
-    skipped, while the fake transport proves `Channel.send()` still
-    transmits the original payload and traceback-hiding flag.
-
-    '''
-    sent: list[tuple[object, bool]] = []
-
-    class FakeTransport:
-        async def send(
-            self,
-            payload: object,
-            hide_tb: bool = False,
-        ) -> None:
-            sent.append((payload, hide_tb))
-
-    def fail_pformat(payload: object) -> str:
-        raise AssertionError(
-            f'suppressed log rendered payload: {payload!r}'
-        )
-
-    chan_log = log.get_logger(
-        name=f'guard_test.{suppression}',
-    )
-    std_log = chan_log.logger
-    orig_level: int = std_log.level
-    orig_disable: int = logging.root.manager.disable
-    transport_level: int = log.CUSTOM_LEVELS['TRANSPORT']
-
-    monkeypatch.setattr(_chan, 'log', chan_log)
-    monkeypatch.setattr(_chan, 'pformat', fail_pformat)
-    try:
-        logging.disable(logging.NOTSET)
-        std_log.setLevel(transport_level)
-
-        if suppression == 'level':
-            std_log.setLevel(logging.INFO)
-        elif suppression == 'logger':
-            monkeypatch.setattr(std_log, 'disabled', True)
-        else:
-            logging.disable(logging.CRITICAL)
-
-        assert not chan_log.isEnabledFor(transport_level)
-
-        transport = FakeTransport()
-        chan = _chan.Channel(transport=transport)
-        payload = object()
-
-        async def send_payload() -> None:
-            await chan.send(
-                payload,
-                hide_tb=True,
-            )
-
-        trio.run(send_payload)
-        assert sent == [(payload, True)]
-    finally:
-        std_log.setLevel(orig_level)
-        logging.disable(orig_disable)
-
-
 # TODO, moar tests against existing feats:
 # ------ - ------
 # - [ ] color settings?

tests/test_resource_cache.py

@@ -9,7 +9,6 @@ from typing import Awaitable
 
 import pytest
 import trio
-from trio.testing import wait_all_tasks_blocked
 import tractor
 from tractor.trionics import (
     maybe_open_context,
@@ -95,232 +94,6 @@ def test_resource_only_entered_once(key_on):
     trio.run(main)
 
 
-def test_last_moc_user_waits_for_resource_exit():
-    '''
-    Verify the final user cannot return before resource teardown.
-
-    Previously the final `maybe_open_context()` user only signalled
-    `_Cache.run_ctx()` through its `no_more_users` event. The user
-    then returned while the service task was still running the
-    resource's `__aexit__()`, so callers could observe stale external
-    state immediately after their `async with` block.
-
-    The resource sets `exit_started` before blocking on
-    `allow_exit`. The user task must remain inside MOC until the test
-    releases that deterministic checkpoint and `__aexit__()` sets
-    `exit_finished`.
-
-    '''
-    async def main():
-        exit_started = trio.Event()
-        allow_exit = trio.Event()
-        exit_finished = trio.Event()
-        user_returned = trio.Event()
-
-        @acm
-        async def open_resource():
-            try:
-                yield
-            finally:
-                exit_started.set()
-                await allow_exit.wait()
-                exit_finished.set()
-
-        async def use_resource():
-            async with maybe_open_context(open_resource):
-                pass
-
-            assert exit_finished.is_set()
-            user_returned.set()
-
-        async with (
-            tractor.open_root_actor(),
-            trio.open_nursery() as tn,
-        ):
-            tn.start_soon(use_resource)
-            await exit_started.wait()
-            assert not user_returned.is_set()
-            allow_exit.set()
-            await user_returned.wait()
-
-    trio.run(main)
-
-
-def test_moc_delivers_resource_exit_error():
-    '''
-    Verify a resource exit error reaches the final MOC user.
-
-    Previously `_Cache.run_ctx()` executed the cached resource's
-    `__aexit__()` after the final user had returned. An exit failure
-    therefore surfaced later through the actor service nursery rather
-    than at the user's `async with maybe_open_context()` boundary.
-
-    This resource raises a unique `ResourceExitError` during exit.
-    Catching that exact instance around MOC proves the service task
-    delivered the failure to the final user without replacing it.
-
-    '''
-    class ResourceExitError(Exception):
-        pass
-
-    exit_error = ResourceExitError('resource exit failed')
-
-    async def main():
-        @acm
-        async def open_resource():
-            yield
-            raise exit_error
-
-        async with tractor.open_root_actor():
-            with pytest.raises(ResourceExitError) as exc_info:
-                async with maybe_open_context(open_resource):
-                    pass
-
-            assert exc_info.value is exit_error
-
-    trio.run(main)
-
-
-def test_moc_final_user_cancellation_waits_for_exit():
-    '''
-    Verify final-user cancellation still waits for successful exit.
-
-    Previously cancellation escaped the final MOC user immediately
-    after it signalled `_Cache.run_ctx()`, leaving resource exit to
-    finish later in the actor service task. This violated the context
-    manager boundary even when cleanup itself succeeded.
-
-    The consumer cancels its own scope while holding the sole cached
-    resource. The resource sets `exit_finished` from its `finally`
-    block, and the consumer checks that event immediately after its
-    cancel scope catches `trio.Cancelled`. This proves MOC's
-    completion wait is shielded without suppressing the original
-    cancellation.
-
-    '''
-    async def main():
-        exit_finished = trio.Event()
-
-        @acm
-        async def open_resource():
-            try:
-                yield
-            finally:
-                exit_finished.set()
-
-        async with tractor.open_root_actor():
-            with trio.CancelScope() as cs:
-                async with maybe_open_context(open_resource):
-                    cs.cancel()
-                    await trio.sleep_forever()
-
-            assert cs.cancelled_caught
-            assert exit_finished.is_set()
-
-    trio.run(main)
-
-
-def test_moc_exit_error_masks_final_user_cancellation():
-    '''
-    Verify cleanup errors survive final-user cancellation.
-
-    A cancelled final user previously signalled `no_more_users` and
-    propagated `trio.Cancelled` before `_Cache.run_ctx()` completed
-    resource exit. If `__aexit__()` then failed, its error was
-    detached from the API call which caused teardown.
-
-    The consumer cancels its own scope at a deterministic checkpoint
-    inside MOC. Resource exit raises `ResourceExitError`; observing
-    that exact error outside the cancel scope proves MOC shields the
-    completion wait and applies normal context-manager masking, where
-    a cleanup failure replaces the active cancellation.
-
-    '''
-    class ResourceExitError(Exception):
-        pass
-
-    exit_error = ResourceExitError('resource exit failed')
-
-    async def main():
-        @acm
-        async def open_resource():
-            yield
-            raise exit_error
-
-        async with tractor.open_root_actor():
-            with pytest.raises(ResourceExitError) as exc_info:
-                with trio.CancelScope() as cs:
-                    async with maybe_open_context(open_resource):
-                        cs.cancel()
-                        await trio.sleep_forever()
-
-            assert exc_info.value is exit_error
-
-    trio.run(main)
-
-
-def test_moc_service_nursery_cancellation_completes_exit():
-    '''
-    Verify service-nursery cancellation cannot strand a final user.
-
-    `_Cache.run_ctx()` and an MOC consumer may share a
-    caller-provided service nursery. Cancelling that nursery
-    interrupts the service task's `no_more_users` wait and the
-    consumer body together. A shielded final-user wait would deadlock
-    if `run_ctx()` failed to publish completion while propagating its
-    own `trio.Cancelled`.
-
-    The outer task waits for resource entry, then cancels the exact
-    nursery containing both tasks. The resource shields one cleanup
-    checkpoint and sets `exit_finished`; observing both it and
-    `service_finished` proves cancellation propagated normally while
-    MOC's completion handshake terminated deterministically.
-
-    '''
-    async def main():
-        resource_entered = trio.Event()
-        exit_finished = trio.Event()
-        service_finished = trio.Event()
-        service_tn: trio.Nursery|None = None
-
-        @acm
-        async def open_resource():
-            try:
-                resource_entered.set()
-                yield
-            finally:
-                with trio.CancelScope(shield=True):
-                    await trio.lowlevel.checkpoint()
-                    exit_finished.set()
-
-        async def use_resource(tn: trio.Nursery):
-            async with maybe_open_context(
-                open_resource,
-                tn=tn,
-            ):
-                await trio.sleep_forever()
-
-        async def run_service():
-            nonlocal service_tn
-
-            async with trio.open_nursery() as tn:
-                service_tn = tn
-                tn.start_soon(use_resource, tn)
-
-            service_finished.set()
-
-        async with trio.open_nursery() as outer_tn:
-            outer_tn.start_soon(run_service)
-            await resource_entered.wait()
-            assert service_tn is not None
-            service_tn.cancel_scope.cancel()
-            await service_finished.wait()
-
-        assert exit_finished.is_set()
-
-    trio.run(main)
-
-
 @tractor.context
 async def streamer(
     ctx: tractor.Context,
@@ -775,52 +548,43 @@ def test_moc_reentry_during_teardown(
     loglevel: str,
 ):
     '''
-    Reproduce re-entry while an identical cached context exits.
+    Reproduce the piker `open_cached_client('kraken')` race:
 
-    - multiple tasks use the same `acm_func` with no kwargs,
-      producing an identical `ctx_key`;
-    - all users leave and the final user starts resource teardown;
-    - `_Cache.run_ctx()` removes the cached value and resource entry
-      before entering the resource's blocking `__aexit__()` body;
-    - a new task attempts to enter that same `ctx_key` during exit;
-    - the per-key lock keeps that entrant queued until exit
-      completes;
-    - the entrant then receives a fresh cache miss and resource.
+    - same `acm_func`, NO kwargs (identical `ctx_key`)
+    - multiple tasks share the cached resource
+    - all users exit -> teardown starts
+    - a NEW task enters during `_Cache.run_ctx.__aexit__`
+    - `values[ctx_key]` is gone (popped in inner finally)
+      but `resources[ctx_key]` still exists (outer finally
+      hasn't run yet bc the acm cleanup has checkpoints)
+    - old code: `assert not resources.get(ctx_key)` FIRES
 
-    Without teardown sharing the registration lock, re-entry could
-    race resource replacement while the prior generation was still
-    exiting. The final user could also return before that exit
-    completed.
-
-    The first resource generation signals `in_aexit` and waits on
-    `allow_aexit`. The re-entry task signals `reentry_started` and
-    blocks inside MOC; only after `wait_all_tasks_blocked()` confirms
-    that ordering does the coordinator release cleanup. The entrant
-    must then receive a fresh cache miss. `first_done` additionally
-    proves the first MOC user observed completed teardown before
-    returning.
+    This models the real-world scenario where `brokerd.kraken`
+    tasks concurrently call `open_cached_client('kraken')`
+    (same `acm_func`, empty kwargs, shared `ctx_key`) and
+    the teardown/re-entry race triggers intermittently.
 
     '''
     async def main():
         in_aexit = trio.Event()
-        allow_aexit = trio.Event()
-        reentry_started = trio.Event()
-        generation: int = 0
 
         @acm
         async def cached_client():
             '''
-            Simulate a no-argument `kraken.api.get_client()`.
+            Simulates `kraken.api.get_client()`:
+            - no params (all callers share one `ctx_key`)
+            - slow-ish cleanup to widen the race window
+              between `values.pop()` and `resources.pop()`
+              inside `_Cache.run_ctx`.
 
             '''
-            nonlocal generation
-
-            generation += 1
-            resource_generation: int = generation
             yield 'the-client'
-            if resource_generation == 1:
-                in_aexit.set()
-                await allow_aexit.wait()
+            # Signal that we're in __aexit__ — at this
+            # point `values` has already been popped by
+            # `run_ctx`'s inner finally, but `resources`
+            # is still alive (outer finally hasn't run).
+            in_aexit.set()
+            await trio.sleep(10)
 
         first_done = trio.Event()
 
@@ -834,25 +598,16 @@ def test_moc_reentry_during_teardown(
         async def reenter_during_teardown():
             '''
             Wait for the acm's `__aexit__` to start (meaning
-            the cached value is no longer available), then re-enter.
+            `values` is popped but `resources` still exists),
+            then re-enter — triggering the assert.
 
             '''
             await in_aexit.wait()
-
-            # Tell the coordinator this task is about to enter MOC.
-            # `Event.set()` is not a checkpoint. Though `async with`
-            # awaits MOC's `__aenter__()`, its async generator runs
-            # synchronously until the held per-key `lock.acquire()`
-            # actually suspends this task.
-            reentry_started.set()
             async with maybe_open_context(
                 cached_client,
             ) as (cache_hit, value):
-                assert not cache_hit
                 assert value == 'the-client'
 
-            await first_done.wait()
-
         with trio.fail_after(5):
             async with (
                 tractor.open_root_actor(
@@ -864,15 +619,5 @@ def test_moc_reentry_during_teardown(
             ):
                 tn.start_soon(use_and_exit)
                 tn.start_soon(reenter_during_teardown)
-                await reentry_started.wait()
-
-                # Wait until the re-entry task is queued on MOC's
-                # per-key lock while `_Cache.run_ctx()` remains
-                # blocked in the first generation's `__aexit__()`.
-                # Only then release cleanup, making the intended
-                # enter-during-sibling-exit ordering deterministic.
-                await wait_all_tasks_blocked()
-                assert not first_done.is_set()
-                allow_aexit.set()
 
     trio.run(main)

tests/trionics/test_taskc.py

@@ -1,333 +0,0 @@
-'''
-`tractor.trionics._taskc.start_or_cancel()` unit tests.
-
-`trio.Nursery.start()` collapses an out-of-band (ancestor)
-cancellation into a lossy,
-
-  `RuntimeError('child exited without calling
-  task_status.started()')`
-
-whenever the started child exits pre-`.started()` WITHOUT
-propagating the ambient `trio.Cancelled`; a common outcome
-when the child (or any lib code it calls) runs a graceful
-teardown which absorbs the cancel and returns early. Our
-`start_or_cancel()` wrapper re-surfaces the real in-flight
-cancellation in that case so the true root error/cancel
-propagates to the `.start()` caller instead.
-
-These tests verify both that repair AND document upstream
-`trio`'s current lossy behaviour via the
-`use_start_or_cancel=False` parametrizations; if a `trio`
-upgrade breaks one of THOSE cases it likely means upstream
-shipped better startup-cancellation porcelain and our
-wrapper deserves a re-audit!
-
-The core use case was dug out of `modden`'s
-`progman.open_wks()` program-spawn machinery as per gh
-issue #474; the wrapper landed originally via gh PR #464.
-
-'''
-import pytest
-import trio
-from trio import TaskStatus
-
-from tractor.trionics import start_or_cancel
-
-
-async def absorbs_cancel_pre_started(
-    task_status: TaskStatus[None] = trio.TASK_STATUS_IGNORED,
-):
-    '''
-    Swallow the ambient (ancestor-scope) cancel and return
-    early, a naughty-but-realistic graceful-teardown pattern
-    and the exact shape which causes `trio.Nursery.start()`
-    to raise its lossy startup `RuntimeError` in place of
-    the real `trio.Cancelled`.
-
-    '''
-    try:
-        await trio.sleep_forever()
-    except trio.Cancelled:
-        return
-
-
-async def raise_val_err():
-    '''
-    Sibling task which blows up (fast) thus OOB-cancelling
-    the shared parent-nursery's cancel-scope.
-
-    '''
-    await trio.lowlevel.checkpoint()
-    raise ValueError('sibling blew up!')
-
-
-@pytest.mark.parametrize(
-    'use_start_or_cancel',
-    [
-        True,
-        False,
-    ],
-)
-def test_sibling_err_not_masked_by_startup_rte(
-    use_start_or_cancel: bool,
-):
-    '''
-    The `modden.runtime.progman` use case: a sibling task
-    errors while the `.start()`-ed child is still
-    pre-`.started()`, OOB-cancelling the shared nursery
-    scope; the child absorbs its cancel (graceful teardown)
-    and exits early.
-
-    - with `start_or_cancel()` the in-flight cancellation
-      is re-surfaced as the real `trio.Cancelled` (then
-      absorbed by the cancelled nursery scope) so ONLY the
-      root-cause sibling error escapes the nursery.
-
-    - with a bare `.start()`, upstream `trio` (currently)
-      also delivers its lossy startup `RuntimeError`
-      alongside, obscuring that the child was in fact
-      cancelled due to the sibling's error.
-
-    `cancelled_at_start` records that the wrapper's own await
-    raises `Cancelled`, rather than merely relying on the
-    nursery's eventual exception-group shape.
-
-    '''
-    cancelled_at_start: list[bool] = []
-
-    async def main():
-        async with trio.open_nursery() as tn:
-            tn.start_soon(raise_val_err)
-            if use_start_or_cancel:
-                try:
-                    await start_or_cancel(
-                        tn,
-                        absorbs_cancel_pre_started,
-                    )
-                except trio.Cancelled:
-                    cancelled_at_start.append(True)
-                    raise
-            else:
-                await tn.start(absorbs_cancel_pre_started)
-
-    with pytest.raises(ExceptionGroup) as excinfo:
-        trio.run(main)
-
-    eg: ExceptionGroup = excinfo.value
-    val_eg, rest_eg = eg.split(ValueError)
-    assert len(val_eg.exceptions) == 1
-
-    if use_start_or_cancel:
-        assert cancelled_at_start == [True]
-        # the re-surfaced `Cancelled` is absorbed by the
-        # (sibling-error cancelled) nursery scope leaving
-        # NO startup-noise, just the root cause.
-        assert rest_eg is None
-    else:
-        # the `trio` wart: a lossy startup RTE rides along
-        # with (and distracts from) the root cause.
-        rte = rest_eg.exceptions[0]
-        assert isinstance(rte, RuntimeError)
-        assert 'child exited without calling' in rte.args[0]
-
-
-@pytest.mark.parametrize(
-    'use_start_or_cancel',
-    [
-        True,
-        False,
-    ],
-)
-def test_pure_oob_cancel_not_morphed_to_rte(
-    use_start_or_cancel: bool,
-):
-    '''
-    A plain (error-free) ancestor `CancelScope.cancel()`
-    fired while the (cancel-absorbing) child is still
-    pre-`.started()`:
-
-    - `start_or_cancel()` re-surfaces the `Cancelled` so
-      the cancelled scope exits CLEAN, no error at all.
-
-    - a bare `.start()` (currently) morphs the plain
-      cancel into an (eg-wrapped) startup `RuntimeError`.
-
-    `cancelled_at_start` proves cancellation interrupts the
-    wrapper call itself before the cancelled scope exits.
-
-    '''
-    cancelled_at_start: list[bool] = []
-
-    async def main():
-        with trio.CancelScope() as cs:
-            async with trio.open_nursery() as tn:
-
-                async def canceller():
-                    await trio.lowlevel.checkpoint()
-                    cs.cancel()
-
-                tn.start_soon(canceller)
-                if use_start_or_cancel:
-                    try:
-                        await start_or_cancel(
-                            tn,
-                            absorbs_cancel_pre_started,
-                        )
-                    except trio.Cancelled:
-                        cancelled_at_start.append(True)
-                        raise
-                else:
-                    await tn.start(
-                        absorbs_cancel_pre_started,
-                    )
-
-        assert cs.cancelled_caught
-
-    if use_start_or_cancel:
-        trio.run(main)
-        assert cancelled_at_start == [True]
-    else:
-        with pytest.raises(ExceptionGroup) as excinfo:
-            trio.run(main)
-
-        rte = excinfo.value.exceptions[0]
-        assert isinstance(rte, RuntimeError)
-        assert 'child exited without calling' in rte.args[0]
-
-
-@pytest.mark.parametrize(
-    'use_start_or_cancel',
-    [
-        True,
-        False,
-    ],
-)
-def test_genuine_startup_rte_still_raised(
-    use_start_or_cancel: bool,
-):
-    '''
-    Absent ANY in-flight cancellation, a child exiting
-    cleanly without calling `task_status.started()` is a
-    genuine startup-protocol bug; `start_or_cancel()` must
-    re-raise the resulting `RuntimeError` exactly like a
-    bare `.start()` does.
-
-    '''
-    async def exits_wo_started(
-        task_status: TaskStatus[None] = (
-            trio.TASK_STATUS_IGNORED
-        ),
-    ):
-        await trio.lowlevel.checkpoint()
-
-    async def main():
-        async with trio.open_nursery() as tn:
-            with pytest.raises(RuntimeError) as excinfo:
-                if use_start_or_cancel:
-                    await start_or_cancel(
-                        tn,
-                        exits_wo_started,
-                    )
-                else:
-                    await tn.start(exits_wo_started)
-
-            rte = excinfo.value
-            assert (
-                'child exited without calling'
-                in
-                rte.args[0]
-            )
-
-    trio.run(main)
-
-
-@pytest.mark.parametrize(
-    'rte_arg',
-    [
-        # Broad substring matches would wrongly demote either
-        # child-owned error to `Cancelled` under cancellation.
-        'never got started!',
-        'child exited without calling user hook',
-        # non-`str` first-arg edge; must not `TypeError`
-        # inside the wrapper's msg-match guard.
-        1234,
-    ],
-)
-def test_childs_own_rte_never_demoted_to_cancel(
-    rte_arg: str|int,
-):
-    '''
-    A child's OWN `RuntimeError`, one which merely smells
-    like `trio`'s startup wording (or carries a non-`str`
-    first arg), raised under ambient cancellation must NOT
-    be demoted to a `trio.Cancelled` by the exact-msg-match
-    guard inside `start_or_cancel()`; the real error must
-    always propagate to the caller as the sole exception-group
-    leaf, preserving object identity.
-
-    '''
-    child_rte = RuntimeError(rte_arg)
-
-    async def cancels_cs_then_raises(
-        task_status: TaskStatus[None] = (
-            trio.TASK_STATUS_IGNORED
-        ),
-    ):
-        # cancel the ambient (ancestor) scope then raise
-        # sync-ly, no checkpoint between, so the child
-        # deterministically dies with ITS error while the
-        # caller is under effective cancellation.
-        cs.cancel()
-        raise child_rte
-
-    cs = trio.CancelScope()
-
-    async def main():
-        with cs:
-            async with trio.open_nursery() as tn:
-                await start_or_cancel(
-                    tn,
-                    cancels_cs_then_raises,
-                )
-
-    with pytest.raises(ExceptionGroup) as excinfo:
-        trio.run(main)
-
-    assert excinfo.value.exceptions == (child_rte,)
-
-
-def test_started_value_and_args_passthru():
-    '''
-    Happy path: positional args, the `name=` kwarg and the
-    `.started(value)`-delivered value all pass through
-    `start_or_cancel()` identically to a bare `.start()`.
-
-    '''
-    async def echo_started(
-        *args,
-        task_status: TaskStatus[tuple] = (
-            trio.TASK_STATUS_IGNORED
-        ),
-    ):
-        task_name: str = trio.lowlevel.current_task().name
-        task_status.started((
-            args,
-            task_name,
-        ))
-
-    async def main():
-        async with trio.open_nursery() as tn:
-            (
-                args,
-                task_name,
-            ) = await start_or_cancel(
-                tn,
-                echo_started,
-                'chillin',
-                10,
-                name='doggy',
-            )
-            assert args == ('chillin', 10)
-            assert task_name == 'doggy'
-
-    trio.run(main)

tractor/ipc/_chan.py

@@ -198,6 +198,9 @@ class Channel:
         # assert transport.raddr == addr
         chan = Channel(transport=transport)
 
+        # ?TODO, compact this into adapter level-methods?
+        # -[ ] would avoid extra repr-calcs if level not active?
+        #   |_ how would the `calc_if_level` look though? func?
         if log.at_least_level('runtime'):
             from tractor.devx import (
                 pformat as _pformat,
@@ -322,12 +325,10 @@ class Channel:
         '''
         __tracebackhide__: bool = hide_tb
         try:
-            if log.at_least_level('transport'):
-                # don't materialize the payload repr if not necessary
-                log.transport(
-                    '=> send IPC msg:\n\n'
-                    f'{pformat(payload)}\n'
-                )
+            log.transport(
+                '=> send IPC msg:\n\n'
+                f'{pformat(payload)}\n'
+            )
             # assert self._transport  # but why typing?
             await self._transport.send(
                 payload,

tractor/ipc/_transport.py

@@ -309,10 +309,7 @@ class MsgpackTransport(MsgTransport):
             log.transport(f'received header {size}')  # type: ignore
             msg_bytes: bytes = await self.recv_stream.receive_exactly(size)
 
-            if log.at_least_level('transport'):
-                log.transport(  # type: ignore
-                    f'received {msg_bytes}'
-                )
+            log.transport(f"received {msg_bytes}")  # type: ignore
             try:
                 # NOTE: lookup the `trio.Task.context`'s var for
                 # the current `MsgCodec`.

tractor/log.py

@@ -112,7 +112,9 @@ def at_least_level(
     if isinstance(level, str):
         level: int = CUSTOM_LEVELS[level.upper()]
 
-    return log.isEnabledFor(level)
+    if log.getEffectiveLevel() <= level:
+        return True
+    return False
 
 
 # TODO, compare with using a "filter" instead?

tractor/msg/_ops.py

@@ -306,15 +306,15 @@ class PldRx(Struct):
             ):
                 try:
                     pld: PayloadT = self._pld_dec.decode(pld)
-                    if log.at_least_level('runtime'):
-                        # don't materialize the payload repr if not necessary
-                        log.runtime(
-                            f'Decoded payload for\n'
-                            f'\n'
-                            f'{msg}\n'
-                            f'where, '
-                            f'{type(msg).__name__}.pld={pld!r}\n'
-                        )
+                    log.runtime(
+                        f'Decoded payload for\n'
+                        # f'\n'
+                        f'{msg}\n'
+                        # ^TODO?, ideally just render with `,
+                        # pld={decode}` in the `msg.pformat()`??
+                        f'where, '
+                        f'{type(msg).__name__}.pld={pld!r}\n'
+                    )
                     return pld
                 except TypeError as typerr:
                     __tracebackhide__: bool = False

tractor/runtime/_rpc.py

@@ -1003,18 +1003,20 @@ async def process_messages(
             task_status.started(loop_cs)
 
             async for msg in chan:
-                if log.at_least_level('transport'):
-                    log.transport(   # type: ignore
-                        f'IPC msg from peer\n'
-                        f'<= {chan.aid.reprol()}\n\n'
+                log.transport(   # type: ignore
+                    f'IPC msg from peer\n'
+                    f'<= {chan.aid.reprol()}\n\n'
 
-                        # TODO: pretty-printing structs is FRAGILE;
-                        # -[ ] add a non-raising log formatter with
-                        #      native-repr fallback before using
-                        #      `.msg.pretty_struct` here.
-                        # f'{pretty_struct.pformat(msg)}\n'
-                        f'{msg}\n'
-                    )
+                    # TODO: use of the pprinting of structs is
+                    # FRAGILE and should prolly not be
+                    #
+                    # avoid fmting depending on loglevel for perf?
+                    # -[ ] specifically `pretty_struct.pformat()` sub-call..?
+                    #   - how to only log-level-aware actually call this?
+                    # -[ ] use `.msg.pretty_struct` here now instead!
+                    # f'{pretty_struct.pformat(msg)}\n'
+                    f'{msg}\n'
+                )
 
                 match msg:
                     # msg for an ongoing IPC ctx session, deliver msg to
@@ -1260,12 +1262,11 @@ async def process_messages(
                         log.exception(message)
                         raise RuntimeError(message)
 
-                if log.at_least_level('transport'):
-                    log.transport(
-                        'Waiting on next IPC msg from\n'
-                        f'peer: {chan.aid.reprol()}\n'
-                        f'|_{chan}\n'
-                    )
+                log.transport(
+                    'Waiting on next IPC msg from\n'
+                    f'peer: {chan.aid.reprol()}\n'
+                    f'|_{chan}\n'
+                )
 
             # END-OF `async for`:
             # IPC disconnected via `trio.EndOfChannel`, likely

tractor/trionics/_mngrs.py

@@ -198,16 +198,6 @@ async def gather_contexts(
 # Further potential examples of interest:
 # https://gist.github.com/njsmith/cf6fc0a97f53865f2c671659c88c1798#file-cache-py-L8
 
-class _CtxExit:
-    '''
-    Completion state for a cached context's shared exit.
-
-    '''
-    def __init__(self) -> None:
-        self.done = trio.Event()
-        self.error: Exception|None = None
-
-
 class _Cache:
     '''
     Globally (actor-processs scoped) cached, task access to
@@ -223,11 +213,7 @@ class _Cache:
     values: dict[Any,  Any] = {}
     resources: dict[
         Hashable,
-        tuple[
-            trio.Nursery,
-            trio.Event,
-            _CtxExit,
-        ],
+        tuple[trio.Nursery, trio.Event]
     ] = {}
     # nurseries: dict[int, trio.Nursery] = {}
     no_more_users: trio.Event|None = None
@@ -237,38 +223,18 @@ class _Cache:
         cls,
         mng,
         ctx_key: tuple,
-        ctx_exit: _CtxExit,
         task_status: trio.TaskStatus[T] = trio.TASK_STATUS_IGNORED,
 
     ) -> None:
-        entered: bool = False
-        try:
-            async with mng as value:
-                entered = True
-                (
-                    _,
-                    no_more_users,
-                    _,
-                ) = cls.resources[ctx_key]
-                cls.values[ctx_key] = value
-                task_status.started(value)
-                try:
-                    await no_more_users.wait()
-                finally:
-                    cls.values.pop(ctx_key)
-                    cls.resources.pop(ctx_key)
-
-        except Exception as exc:
-            if not entered:
-                raise
-
-            # Deliver regular `__aexit__()` failures to the final
-            # consumer instead of raising into the service nursery.
-            ctx_exit.error = exc
-
-        finally:
-            if entered:
-                ctx_exit.done.set()
+        async with mng as value:
+            _, no_more_users = cls.resources[ctx_key]
+            cls.values[ctx_key] = value
+            task_status.started(value)
+            try:
+                await no_more_users.wait()
+            finally:
+                value = cls.values.pop(ctx_key)
+                cls.resources.pop(ctx_key)
 
 
 class _UnresolvedCtx:
@@ -315,10 +281,9 @@ async def maybe_open_context(
         )
 
     # yielded output
+    # sentinel = object()
     yielded: Any = _UnresolvedCtx
     user_registered: bool = False
-    ctx_exit: _CtxExit|None = None
-    exit_error: Exception|None = None
 
     # Lock resource acquisition around task racing  / ``trio``'s
     # scheduler protocol.
@@ -335,6 +300,7 @@ async def maybe_open_context(
         ] = trio.StrictFIFOLock()
         header: str = 'Allocated NEW lock for @acm_func,\n'
     else:
+        await trio.lowlevel.checkpoint()
         header: str = 'Reusing OLD lock for @acm_func,\n'
 
     log.debug(
@@ -402,6 +368,7 @@ async def maybe_open_context(
             resources = _Cache.resources
             entry: tuple|None = resources.get(ctx_key)
             if entry:
+                service_tn, ev = entry
                 raise RuntimeError(
                     f'Caching resources ALREADY exist?!\n'
                     f'ctx_key={ctx_key!r}\n'
@@ -409,18 +376,12 @@ async def maybe_open_context(
                     f'task: {task}\n'
                 )
 
-            ctx_exit = _CtxExit()
-            resources[ctx_key] = (
-                service_tn,
-                trio.Event(),
-                ctx_exit,
-            )
+            resources[ctx_key] = (service_tn, trio.Event())
             try:
                 yielded: Any = await service_tn.start(
                     _Cache.run_ctx,
                     mngr,
                     ctx_key,
-                    ctx_exit,
                 )
             except BaseException:
                 # If `run_ctx` (wrapping the acm's `__aenter__`)
@@ -466,11 +427,6 @@ async def maybe_open_context(
             raise taskc
     else:
         # XXX, cached-entry-path
-        (
-            _,
-            _,
-            ctx_exit,
-        ) = _Cache.resources[ctx_key]
         _Cache.users[ctx_key] += 1
         user_registered = True
         log.debug(
@@ -489,57 +445,44 @@ async def maybe_open_context(
         )
 
     finally:
+        if lock.locked():
+            stats: trio.LockStatistics = lock.statistics()
+            owner: trio.Task|None = stats.owner
+            log.error(
+                f'Lock never released by last owner={owner!r} !?\n'
+                f'{stats}\n'
+                f'\n'
+                f'task={task!r}\n'
+                f'ctx_key={ctx_key!r}\n'
+                f'acm_func={acm_func}\n'
+
+            )
+
         if user_registered:
-            # Serialize user registration and teardown under the same
-            # per-key lock so no entrant can acquire a resource after
-            # its final user has committed to exiting it.
-            with trio.CancelScope(shield=True):
-                await lock.acquire()
-                try:
-                    _Cache.users[ctx_key] -= 1
+            _Cache.users[ctx_key] -= 1
 
-                    # If no consumers remain, keep entrants queued
-                    # until the cached context has completely exited.
-                    if _Cache.users[ctx_key] <= 0:
-                        log.debug(
-                            f'De-allocating @acm-func entry\n'
-                            f'ctx_key={ctx_key!r}\n'
-                            f'acm_func={acm_func!r}\n'
-                        )
+        if yielded is not _UnresolvedCtx:
+            # if no more consumers, teardown the client
+            if _Cache.users[ctx_key] <= 0:
+                log.debug(
+                    f'De-allocating @acm-func entry\n'
+                    f'ctx_key={ctx_key!r}\n'
+                    f'acm_func={acm_func!r}\n'
+                )
 
-                        # XXX: if we're cancelled, the entry may
-                        # have never been entered since the nursery
-                        # task was killed.
-                        entry = _Cache.resources.get(ctx_key)
-                        if entry:
-                            (
-                                _,
-                                no_more_users,
-                                ctx_exit,
-                            ) = entry
-                            no_more_users.set()
+                # XXX: if we're cancelled we the entry may have never
+                # been entered since the nursery task was killed.
+                # _, no_more_users = _Cache.resources[ctx_key]
+                entry = _Cache.resources.get(ctx_key)
+                if entry:
+                    _, no_more_users = entry
+                    no_more_users.set()
 
-                        assert ctx_exit is not None
-                        await ctx_exit.done.wait()
-                        exit_error = ctx_exit.error
-
-                        # A queued entrant already holds a reference
-                        # to this lock. Keep it registered until that
-                        # task has acquired and released it.
-                        stats = lock.statistics()
-                        if not stats.tasks_waiting:
-                            maybe_lock = _Cache.locks.get(ctx_key)
-                            if maybe_lock is lock:
-                                _Cache.locks.pop(ctx_key)
-                            else:
-                                log.error(
-                                    f'Resource lock for {ctx_key} '
-                                    f'was replaced before teardown?'
-                                )
-                finally:
-                    lock.release()
-
-        if exit_error is not None:
-            # Always re-raise a regular `__aexit__()` error at the
-            # final consumer's context boundary.
-            raise exit_error
+                maybe_lock = _Cache.locks.pop(
+                    ctx_key,
+                    None,
+                )
+                if maybe_lock is None:
+                    log.error(
+                        f'Resource lock for {ctx_key} ALREADY POPPED?'
+                    )

tractor/trionics/_taskc.py

@@ -349,10 +349,7 @@ async def start_or_cancel(
             # demote it to a `Cancelled`, losing the real error. The
             # `isinstance` guard also avoids a `TypeError` when
             # `rte.args[0]` isn't a `str`.
-            rte.args[0] == (
-                'child exited without calling '
-                'task_status.started()'
-            )
+            'child exited without calling' in rte.args[0]
         ):
             # re-raises the in-flight `trio.Cancelled` IFF we're
             # under effective cancellation; else a cheap no-op and

uv.lock

@@ -900,11 +900,11 @@ wheels = [
 
 [[package]]
 name = "setuptools"
-version = "83.0.0"
+version = "82.0.1"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/34/26/f5d29e25ffdb535afef2d35cdb55b325298f96debd670da4c325e08d70f4/setuptools-83.0.0.tar.gz", hash = "sha256:025bccbbf0fa05b6192bc64ae1e7b16e001fd6d6d4d5de03c97b1c1ade523bef", size = 1154254, upload-time = "2026-07-04T15:31:22.699Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/4f/db/cfac1baf10650ab4d1c111714410d2fbb77ac5a616db26775db562c8fab2/setuptools-82.0.1.tar.gz", hash = "sha256:7d872682c5d01cfde07da7bccc7b65469d3dca203318515ada1de5eda35efbf9", size = 1152316, upload-time = "2026-03-09T12:47:17.221Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/5d/40/e1e72872c6354b306daef1703549e8e83b4d43cfea356311bf722a043752/setuptools-83.0.0-py3-none-any.whl", hash = "sha256:29b23c360f22f414dc7336bb39178cc7bcbf6021ed2733cde173f09dba19abb3", size = 1008090, upload-time = "2026-07-04T15:31:20.885Z" },
+    { url = "https://files.pythonhosted.org/packages/9d/76/f789f7a86709c6b087c5a2f52f911838cad707cc613162401badc665acfe/setuptools-82.0.1-py3-none-any.whl", hash = "sha256:a59e362652f08dcd477c78bb6e7bd9d80a7995bc73ce773050228a348ce2e5bb", size = 1006223, upload-time = "2026-03-09T12:47:15.026Z" },
 ]
 
 [[package]]