to_asyncio: fix `trio`-side-exit handling

In `.open_channel_from()`, when the `trio`-side parent task exits, it
signals the running `asyncio`-side task with `TrioTaskExited`; this is
our internal exit signal to any still-running `asyncio` task. When the
`asyncio`-side raises the issued signal, previously
`translate_aio_errors()` let it escape despite
`suppress_graceful_exits=True`, (in `piker`'s case) turning an expected
(and cached via `.trionics.maybe_open_context`) channel teardown into
a (quote-feed) task-restart failure.

Now if that task raises the same signal, `translate_aio_errors()`
suppresses it when `suppress_graceful_exits=True`, while preserving any
original `trio`-side error. There are two cooperating fixes:

1. `translate_aio_errors()` returns early without raising that TTE. This
   prevents normal channel teardown from propagating it back up to
   the originating task.

2. The `asyncio` done callback `signal_trio_when_done()` now returns
   early for that same TTE ack and prevents it from
   recancelling the already-unwinding `trio-side` or raising the signal
   again as an unhandled callback error.

Deats,
- record the issued `TrioTaskExited` instance in
  `LinkedTaskChannel._aio_owner_exit` before signalling the
  `asyncio`-side task, retaining the original `trio`-side error
  as its cause when present.
- use exception identity (`is`), rather than type, to recognize
  that issued signal; a separately raised TTE or another
  `asyncio`-side cleanup error retains existing error handling.
- in `translate_aio_errors()`, require the `trio`-side parent
  task to have exited before handling the acknowledged signal:
  - raise any original `trio_err` first, without making its own
    exit signal its cause and creating an exception-cause cycle;
  - OW return early under `suppress_graceful_exits=True`;
  - with suppression disabled, raise the exact issued TTE.
- in `signal_trio_when_done()`, retrieve the task exception but
  skip parent-task recancellation and callback error relay when
  it is that issued TTE and the `trio`-side has already exited.
- in `wait_on_coro_final_result()`, retain the issued TTE in
  `chan._aio_err` for the reporting decision above, logging its
  receipt as an exit acknowledgement rather than a task failure.
- retain the actual `maybe_signal_aio_task()` delivery report
  for both acknowledgement logs; include the `trio`-side parent
  task via `pformat_task()`, its original error/cancellation cause,
  and runtime-level diagnostics for each reporting decision.

Update docs/reporting,
- document the signal writer, readers and teardown outcomes;
- retain the actual signalling report for both acknowledgements;
- use `pformat_task()` to identify the `trio.Task` in diagnostics;
- distinguish the parent's error from its issued TTE;
- log suppression, strict reporting and `trio`-side-error precedence;
- clarify graceful suppression for cached-channel teardown;
- keep the queue-shutdown TODO beside its existing handler.

Qualification using a separate, uncommitted regression module:
- original matrix 8 failed / 4 passed; candidate matrix 12 passed.
- selected bridge/formatter run: 41 passed / 18 deselected.
- earlier `piker` real-channel integration: 4 passed.
- human reset traces acknowledge owner exit and preserve a quote
  `TimeoutError`.

Prompt-IO: ai/prompt-io/20261028T193020Z_b64c89f0_cached_owner.md

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

ai/prompt-io/20261028T193020Z_b64c89f0_cached_owner.md

@@ -0,0 +1,140 @@
+---
+created_at: 2026-10-28T19:30:20Z
+generated_by:
+  harness: opencode
+  provider: openai
+  model: openai/gpt-6-astra
+git_ref: b64c89f0
+scope: code
+substantive: true
+raw_file: 20261028T193020Z_b64c89f0_cached_owner.raw.md
+---
+
+## Prompt and scope
+
+The human authorized investigation in an isolated Tractor worktree
+after cherry-picking Piker's original 4af885d9 onto the in-use
+e929fdc2 baseline (replay eb0514c3). They asked to verify related
+connectivity work, challenged calling this a Tractor regression,
+and then said to investigate the issue for real.
+
+The separate worktree is wkt/cached_aio_owner_exit, rooted at
+b64c89f05a9fc8ac2878551eb5eda36574ebef6a. Installed environments and
+the original Piker ownership patch are preserved. No verified dialog
+ID was available, so dialog/worktree association was not recorded.
+
+## Live evidence
+
+On October 28, QQQ backfill hit IB pacing error 162 at 15:02:57.
+Farm reset proceeded to recovery messages at 15:02:59--15:03:00.
+QQQ and MNQ quote consumers closed their cached owners to restart.
+_Cache.run_ctx injected TrioTaskExited into producer tasks 23 and 11.
+The exit signal escaped through translate_aio_errors() and the cache
+into stream_quotes(), entering PDB and cancelling the daemon tree.
+
+The final root error after debugger continuation still named the
+datad.ib TrioTaskExited. Broken pipes, late cancel acknowledgements,
+and EMS ContextCancelled groups followed the initial failure. This
+is an operational reset failure rather than ordinary SIGINT noise.
+
+The evaluated Piker history already contains socket-disconnect
+propagation efa72980 and bounded reset coordination 0846cbd4. Its
+missing timer/sampler changes d126302f, c0dae89a and a26e060c do not
+alter this bridge exit path. They remain separate integration work.
+
+## Contract and implementation
+
+docs/guide/asyncio.rst already states that the default
+suppress_graceful_exits=True absorbs TaskExited signals. A reduced
+Tractor-only test reproduces the violation without IB or Piker.
+No historical first-bad Tractor commit has been established.
+
+LinkedTaskChannel._aio_owner_exit records the exact object issued
+by open_channel_from() at owner exit. The done callback consumes
+that echo instead of cancelling its owner and raising it again.
+The translator preserves genuine Trio errors, suppresses normal
+echoes by default, and reports the same object under explicit policy.
+Type-only suppression is deliberately avoided; cleanup errors and
+newly constructed TrioTaskExited instances retain propagation.
+
+## Verification
+
+Tractor environment: existing /home/goodboy/repos/tractor/.venv,
+pytest 9.1.0, Trio 0.33.0, Greenback 1.2.1, msgspec 0.21.1.
+TRACTOR_SPAWN_METHOD and TRACTOR_LOGLEVEL were unset. PYTHONPATH
+selected only this worktree. Greenback required the existing Nix
+GCC library directory in LD_LIBRARY_PATH; no packages were installed.
+
+Initial development runs exposed two test-environment/harness issues:
+successful RPCs left the test actor running until its nursery timed
+out, and Greenback could not load libstdc++. Those were corrected
+in the test shutdown and process-local environment respectively.
+They are not evidence of additional production runtime failures.
+
+- Original pin + corrected new test: 8 failures / 4 passes.
+  Normal owner exits fail with the echoed signal; strict reporting
+  also fails its exact-identity assertion. Independent-error controls
+  pass. Both native and Greenback paths are exercised.
+- Fixed new matrix: 12 passes, including overlapping cached users
+  in distinct Trio tasks, two generations and two RPCs per case.
+- New matrix + root infection + selected infected-asyncio tests:
+  36 passes, 18 deselected, 24 multiprocessing deprecation warnings.
+  SIGINT and interactive debugger selections were excluded; those
+  paths are not claimed qualified by this run.
+- Frozen Piker e929fdc2 + original 4af885d9 + existing real bridge
+  reproducer, with the duplicate --ll compatibility removal:
+  4 passes with candidate Tractor imported via process-local paths.
+  Both actor survival and strict callback cleanliness now pass.
+  External IB and VNC inputs in that reproducer are replayed; its
+  actor, cache, bridge, event delivery and lifetime boundaries are real.
+- The human advanced the evaluation worktree during investigation to
+  sampler_tick_time_guard at 790dc0a51c9d93e77258360f954c959d11d09a7c.
+  Repeated the same four Piker checks on that exact source tip with
+  candidate Tractor and the same test-only harness adaptation:
+  4 passed. Its working tree/index were preserved, and its installed
+  to_asyncio.py still has the original b64c89f0 blob fingerprint.
+
+The Piker integration used its existing py313 environment and pinned
+native dependency paths. Import probes verified both source roots.
+Raw baseline/integration output remains in the session scratch files
+cached-owner-baseline-final.log and cached-owner-piker-final.log.
+
+The human subsequently reported that a backfill data-connection reset
+no longer crashed at the previous owner-exit failure. Supplied live
+traces resolve Tractor to this worktree and show both asyncio task
+and completion acknowledgements of the issued owner-exit signal.
+
+At 16:29:07, a separate initial IB get_quote() timeout survived proxy
+teardown with both owner-exit acknowledgements. After the human
+continued PDB at 16:31:40, the chart received that TimeoutError at
+16:31:41. IB also reported competing-session error 10197 for MNQ
+request 822; its identity with the timed-out request is unverified.
+
+This is bounded live evidence, not complete recovery qualification.
+Separate unresolved observations include final-result send_nowait()
+raising BrokenResourceError and a callback bookkeeping assertion;
+samplerd result delivery on a closed socket; cancellation-only
+exception groups relayed as errors on SIGINT; Qt self-SIGINT during
+close; and Piker EMS request loss followed by a req=None fill crash.
+Sampler timestamp warnings and volume spikes also remain separate.
+
+Live connection, TSDB, VNC, account and order operations were performed
+only by the human. The agent misread a request for a commit plan and
+created commit 6142d52efc0e5797a240eba0c0b00a24b211fe0e. The human
+clarified that review must precede committing and authorized a soft
+reset to b64c89f05a9fc8ac2878551eb5eda36574ebef6a. All five files
+were preserved staged; the replacement plan is human-executed.
+
+## Files changed
+
+- tractor/to_asyncio.py: exact owner-exit echo classification.
+- tests/test_cached_aio_owner_exit.py: real contract regressions.
+- docs/guide/asyncio.rst: explain ownership-specific suppression.
+- This provenance pair.
+
+## Human edits
+
+The human supplied live failures and teardown traces, required reuse
+inspection of connectivity branches, and corrected the distinction
+between a contract defect and an established historical regression.
+No direct manual edits to this new Tractor worktree were observed.

ai/prompt-io/20261028T193020Z_b64c89f0_cached_owner.raw.md

@@ -0,0 +1,72 @@
+---
+created_at: 2026-10-28T19:30:20Z
+generated_by:
+  harness: opencode
+  provider: openai
+  model: openai/gpt-6-astra
+git_ref: b64c89f0
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Investigated the original Piker IB quote-cache patch on the qualified
+Tractor pin. Normal final-owner closure injects TrioTaskExited into
+the asyncio peer. Its echo was then treated as an independent error
+both by the done callback and by translate_aio_errors().
+
+> `git diff HEAD~1..HEAD -- tractor/to_asyncio.py`
+
+Record the precise owner-exit exception on LinkedTaskChannel. Consume
+that object's completion callback without cancelling an already
+unwinding owner. At the translator boundary preserve the actual
+Trio-side error, absorb a graceful echo under the default policy,
+or raise the exact signal when explicit reporting is requested.
+Independent errors, including newly constructed same-type errors,
+retain existing propagation. Classify the expected echo in logs as
+exit acknowledgement rather than an independent producer failure.
+
+> `git diff HEAD~1..HEAD -- tests/test_cached_aio_owner_exit.py`
+
+Use an infected actor with real bridge/cache/broadcast primitives,
+overlapping consumers, repeated owner retirement/reopening and two
+RPCs on the same actor. Test native and Greenback producer tasks,
+cancelled final consumers, explicit exit reporting, and independent
+RuntimeError/same-type cleanup failures with exception identity and
+cause assertions. Capture asyncio callbacks to reject duplicate
+unhandled echoes. Explicitly cancel the still-running test actor
+after its successful RPCs, before closing its actor nursery.
+
+> `git diff HEAD~1..HEAD -- docs/guide/asyncio.rst`
+
+Clarify the existing default graceful-exit contract: only the exact
+bridge-issued owner-exit signal is eligible for suppression.
+
+Corrected baseline: 8 failed, 4 passed. Fixed matrix: 12 passed.
+Selected Tractor coverage: 36 passed, 18 deselected, with 24 existing
+multiprocessing fork-in-multithreaded-process deprecation warnings.
+Piker's original cached quote patch plus real bridge tests: 4 passed,
+including actor survival and callback cleanliness.
+
+This establishes a defect against the documented contract at the
+pin, not a bisected historical regression between Tractor revisions.
+No installed runtime, Piker source branch or live broker was changed.
+
+## Subsequent human evidence and authorization
+
+The human reported successful passage through the previous backfill
+reset crash. Live logs show the candidate's two owner-exit
+acknowledgements. A later get_quote() TimeoutError remained primary
+through that cleanup, and reached the chart after PDB continuation.
+Separate completion-delivery, SIGINT, Qt, EMS and sampler failures
+remain outside this patch; see the paired structured record.
+
+Human request:
+> so let's get the tractor wkt committed so we can pin there and then
+> move forward with fixin all this new shiz
+
+Human correction:
+> i meant do a commit plan dawg Bp
+
+After the agent offered a one-commit soft reset preserving the files:
+> yeah, bc i want to review before the commit
+
+The authorized soft reset restored b64c89f0 with all files staged.

docs/guide/asyncio.rst

@@ -279,6 +279,13 @@ absorbs the two ``*TaskExited`` signals so happy-path teardown
 stays silent; pass ``False`` when your app wants to handle early
 peer-exit explicitly.
 
+Suppression applies to the exact owner-exit signal issued by the
+channel, not to every exception of the same type. A producer's own
+cleanup exception still propagates, even if it is a newly created
+``TrioTaskExited``. This also applies when ``maybe_open_context()``
+owns the channel: the final consumer releases that resource through
+normal owner exit, including when that consumer was cancelled.
+
 Past the task pair, everything composes with the normal actor
 story: an unhandled ``asyncio`` error is translated into the
 ``trio`` side, propagates out of your ``@tractor.context``

tractor/to_asyncio.py

@@ -50,6 +50,7 @@ from tractor.runtime._state import (
 )
 from tractor._context import Unresolved
 from tractor import devx
+from tractor.devx.pformat import pformat_task
 from tractor.log import (
     get_logger,
     StackLevelAdapter,
@@ -151,6 +152,33 @@ class LinkedTaskChannel(
     # set after `asyncio.create_task()`
     _aio_task: asyncio.Task|None = None
     _aio_err: BaseException|None = None
+    # Identity of OUR shutdown signal, not any TrioTaskExited raised
+    # independently by the producer. open_channel_from() sets this
+    # in its finally block BEFORE maybe_signal_aio_task(), iff the
+    # aio task is still running. This covers normal owner exit,
+    # owner error/cancellation, and last-user cached-context exit.
+    # An already completed aio task leaves this as None; run_task()
+    # does not set it. The signal retains the in-flight owner's
+    # exception as __cause__, including trio.Cancelled when present.
+    #
+    # Readers identify the original issued signal by `is`:
+    # - wait_on_coro_final_result() records it in _aio_err and logs
+    #   an acknowledgement instead of an independent producer fault.
+    # - signal_trio_when_done() retrieves the task exception, logs
+    #   its acknowledgement and, with _trio_exited, returns without
+    #   recancelling the owner or raising a second callback failure.
+    # - translate_aio_errors(), with _trio_exited, raises a captured
+    #   trio_err first. Otherwise suppress_graceful_exits=True
+    #   absorbs the acknowledged signal; False raises that object.
+    #   Raising trio_err bare avoids making its signal its cause.
+    #
+    # A different TrioTaskExited or other producer error does not
+    # match: existing independent-error/relay-cause handling applies.
+    # This field never authorizes blanket suppression by exc type.
+    _aio_owner_exit: TrioTaskExited|None = None
+    # Actual delivery report from maybe_signal_aio_task(), retained
+    # for both acknowledgement sites; diagnostic only, not policy.
+    _aio_owner_exit_report: str|None = None
     _aio_to_raise: (
         TrioTaskExited|  # trio task exits while aio ongoing
         BaseException|
@@ -479,6 +507,15 @@ def _run_asyncio_task(
                 log.runtime(
                     '`asyncio` task was cancelled..\n'
                 )
+            elif aio_err is chan._aio_owner_exit:
+                owner: str = pformat_task(chan._trio_task)
+                log.cancel(
+                    f'Asyncio task acknowledged owner exit\n'
+                    f'owner:\n{owner}\n'
+                    f'owner error/cancel: {aio_err.__cause__!r}\n'
+                    f'{chan._aio_owner_exit_report}\n'
+                )
+
             else:
                 log.exception(
                     '`asyncio` task errored\n'
@@ -681,6 +718,15 @@ def _run_asyncio_task(
                     msg.format(etype_str='cancelled')
                 )
 
+            elif aio_err is chan._aio_owner_exit:
+                owner: str = pformat_task(chan._trio_task)
+                log.cancel(
+                    f'Asyncio completion acknowledged exit\n'
+                    f'owner:\n{owner}\n'
+                    f'owner error/cancel: {aio_err.__cause__!r}\n'
+                    f'{chan._aio_owner_exit_report}\n'
+                )
+
             # XXX when the asyncio.Task exits early (before the trio
             # side) we relay through an exc-as-signal which is
             # normally suppressed unless the trio.Task also errors
@@ -718,6 +764,25 @@ def _run_asyncio_task(
         # is trio the src of the aio task's exc-as-outcome?
         trio_err: BaseException|None = chan._trio_err
         curr_aio_err: BaseException|None = chan._aio_err
+        if (
+            chan._trio_exited
+            and aio_err is not None
+            and aio_err is chan._aio_owner_exit
+        ):
+            # The owner is already unwinding and waiting for this
+            # task. Let translate_aio_errors() apply its reporting
+            # policy; cancelling the owner or raising here
+            # would report our exit signal as a second failure.
+            owner: str = pformat_task(chan._trio_task)
+            log.runtime(
+                f'Owner-exit acknowledged; skipping recancellation\n'
+                f'and callback error relay\n'
+                f'translate_aio_errors() decides what to raise\n'
+                f'owner:\n{owner}\n'
+                f'owner-exit signal: {aio_err!r}\n'
+            )
+            return
+
         if (
             curr_aio_err
             or
@@ -907,7 +972,7 @@ def maybe_signal_aio_task(
     See the "cross-loop cause-chain matrix" comment in
     `translate_aio_errors()`'s final-raise block for how this
     `cause` interacts with every `raise X [from Y]` exit path
-    (esp. the relay-echo guard which prevents a cause CYCLE).
+    (especially the relay-signal guard preventing a cause cycle).
 
     '''
     if cause is not None and exc.__cause__ is None:
@@ -1337,7 +1402,7 @@ async def translate_aio_errors(
         #                     relay exc (`AsyncioCancelled`/`AsyncioTaskExited`).
         # - raise_from      : `trio_err if (aio_err is trio_to_raise)
         #                      else aio_err` (the chosen `__cause__`).
-        # - relay-echo      : an `aio_err` that is one of OUR OWN
+        # - relay-signal    : an `aio_err` that is one of OUR OWN
         #                     `TrioTaskExited|TrioCancelled` signals,
         #                     synth'd + delivered to the aio-side by
         #                     `maybe_signal_aio_task()`; its `__cause__`
@@ -1347,12 +1412,15 @@ async def translate_aio_errors(
         # this block (final-raise in `translate_aio_errors`):
         #   condition                            =>  raises         from
         #   -----------------------------------      -------------  -----------
+        #   owner-exit acknowledgement + trio_err => trio_err (bare)
+        #   owner-exit acknowledgement, suppressed => return
+        #   owner-exit acknowledgement, strict => aio_err (bare)
         #   not suppress_graceful_exits          =>  trio_to_raise  raise_from
         #   AsyncioTaskExited + trio Cancelled/None  => return (aio-exit ignored)
         #   AsyncioTaskExited + trio EoC         =>  trio_err       (bare)
         #   AsyncioCancelled  + trio Cancelled   =>  return (co-cancel ignored)
         #   trio_to_raise match catch-all        =>  trio_to_raise  raise_from
-        #   aio_err is relay-echo  ◄── the GUARD  =>  trio_err       (bare)
+        #   aio_err is our relay signal => trio_err (bare)
         #   aio_err independent (real aio fail)  =>  trio_err       aio_err
         #   aio_err independent, no trio_err     =>  aio_err        (bare)
         #   only trio_err                        =>  trio_err       (bare)
@@ -1361,7 +1429,7 @@ async def translate_aio_errors(
         #   AsyncioTaskExited relay-out          =>  trio_to_raise  aio_err
         #   plain aio_err re-raise               =>  aio_err  (__cause__ preset)
         #
-        # INVARIANT: a relay-echo must NEVER become `trio_err.__cause__`
+        # Our relay signal must NEVER become `trio_err.__cause__`
         #  (it's ALREADY caused-BY `trio_err`) → doing so would CYCLE
         #  (`trio_err ◄─► relay`). So the guard raises the root
         #  `trio_err` bare; the relay still keeps its own correct
@@ -1380,6 +1448,37 @@ async def translate_aio_errors(
             else aio_err
         )
 
+        if (
+            chan._trio_exited
+            and aio_err is not None
+            and aio_err is chan._aio_owner_exit
+        ):
+            owner: str = pformat_task(chan._trio_task)
+            if trio_err is not None:
+                log.runtime(
+                    f'Preserving Trio error after owner exit\n'
+                    f'owner:\n{owner}\n'
+                    f'trio_err: {trio_err!r}\n'
+                )
+                raise trio_err
+
+            if suppress_graceful_exits:
+                log.runtime(
+                    f'Suppressing acknowledged owner-exit signal\n'
+                    f'suppress_graceful_exits=True, trio_err=None\n'
+                    f'owner:\n{owner}\n'
+                    f'owner-exit signal: {aio_err!r}\n'
+                )
+                return
+
+            log.runtime(
+                f'Raising acknowledged owner-exit signal\n'
+                f'suppress_graceful_exits=False, trio_err=None\n'
+                f'owner:\n{owner}\n'
+                f'owner-exit signal: {aio_err!r}\n'
+            )
+            raise aio_err
+
         if not suppress_graceful_exits:
             raise trio_to_raise from raise_from
 
@@ -1427,8 +1526,8 @@ async def translate_aio_errors(
             # (`TrioTaskExited`/`TrioCancelled`) that we delivered
             # to the aio-side via `maybe_signal_aio_task()`, AND
             # its `__cause__` already points back at `trio_err`,
-            # then it's just a derivative ECHO of the trio-side
+            # then it is our relay signal for the Trio-side error,
-            # error, NOT an independent asyncio failure.
+            # NOT an independent asyncio failure.
             #
             # Raising `trio_err from aio_err` here would invert
             # (and cyclically tangle) the cause chain since the
@@ -1571,16 +1670,28 @@ async def open_channel_from(
                     # `sys.exc_info()[1]` is non-`None` only when
                     # the `try` body raised (graceful exit -> None).
                     trio_exc: BaseException|None = sys.exc_info()[1]
+                    owner_exit: TrioTaskExited = TrioTaskExited(
+                        f'but the child `asyncio` task is still '
+                        f'running?\n'
+                        f'>>\n'
+                        f' |_{aio_task!r}\n'
+                    )
+                    chan._aio_owner_exit = owner_exit
                     _, report = maybe_signal_aio_task(
                         aio_task,
-                        TrioTaskExited(
+                        owner_exit,
-                            f'but the child `asyncio` task is still running?\n'
-                            f'>>\n'
-                            f' |_{aio_task!r}\n'
-                        ),
                         cause=trio_exc,
                     )
-                    log.cancel(report)
+                    chan._aio_owner_exit_report = report
+                    owner: str = pformat_task(chan._trio_task)
+                    log.cancel(
+                        f'Trio owner exited with aio task ongoing\n'
+                        f'owner:\n{owner}\n'
+                        f'owner error/cancel: {trio_exc!r}\n'
+                        f'bridge cancel_called: {cs.cancel_called}\n'
+                        f'{report}\n'
+                    )
+
                 else:
                     chan._to_trio.close()