Bound cancelled remote-task startup

Cancellation after `Start` publication but before `StartAck` can
strand the caller context and leave its remote task running.

Make one shielded, bounded task-cancel request before dropping
local startup state. Keep the private `cancel_on_startup` policy
outside public target kwargs and disable it for the `_cancel_task`
RPC itself so cleanup can not recursively cancel its own startup.

Release each private helper context on exit and prove the
caller-owned actor remains reusable after controlled startup
cancellation.

Caught-during: review remediation
Found-via: `/run-tests` test_cancel_during_context_startup

Review: PR #481 (opencode)
https://github.com/goodboy/tractor/pull/481#pullrequestreview-4956692120

Prompt-IO: ai/prompt-io/opencode/20260818T193003Z_bf06b4f8_prompt_io.md

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

ai/prompt-io/opencode/20260818T193003Z_bf06b4f8_prompt_io.md

@@ -0,0 +1,38 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_3e4c90d3eafeqHEtRYSIHgHhpA
+timestamp: 2026-08-18T19:30:03Z
+git_ref: bf06b4f8
+scope: code
+substantive: true
+raw_file: 20260818T193003Z_bf06b4f8_prompt_io.raw.md
+---
+
+## Prompt
+
+Cancel a remote task when its caller is cancelled after `Start`
+publication but before startup acknowledgement. Keep cancellation
+bounded, prevent its private `_cancel_task` RPC from recursively
+cancelling itself and preserve public target kwargs unchanged.
+
+## Response summary
+
+Add private portal startup policy, use it for non-recursive context
+cancellation and clean caller-side startup state under a shield.
+
+## Files changed
+
+- `tractor/runtime/_portal.py` - separate private startup policy.
+- `tractor/_context.py` - disable recursion for cancellation RPCs.
+- `tractor/runtime/_runtime.py` - clean cancelled task startup.
+- `tests/test_context_stream_semantics.py` - control cancellation
+  between `Start` publication and acknowledgement.
+
+## Human edits
+
+The human required this cancellation behavior to remain a distinct
+commit from general startup failures and from the public `to_actor`
+API. The human also requested that its runtime comment describe the
+actual length-prefixed transport guarantee and concrete `_cancel_task`
+operation rather than referring to an unnamed wrapper.

ai/prompt-io/opencode/20260818T193003Z_bf06b4f8_prompt_io.raw.md

@@ -0,0 +1,20 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-18T19:30:03Z
+git_ref: bf06b4f8
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Cancellation while `Actor.start_remote_task()` waits for `StartAck`
+can strand its caller-side context and leave the remote task running.
+Make one bounded cleanup request, remove local startup state and close
+its receive channel.
+
+> `git diff HEAD~1..HEAD -- tractor/runtime/_runtime.py tractor/runtime/_portal.py tractor/_context.py tests/test_context_stream_semantics.py`
+
+Separate private startup-cancellation policy from public target kwargs
+using `Portal._run_from_ns()`. Have `Context.cancel()` disable recursive
+startup cancellation for its own `_cancel_task` RPC. Exercise
+cancellation after `Start` publication and prove the caller-owned actor
+remains reusable without leaked contexts.

tests/test_context_stream_semantics.py

@@ -7,6 +7,7 @@ sync-opening a ``tractor.Context`` beforehand.
 '''
 from itertools import count
 import math
+from pathlib import Path
 import platform
 from pprint import pformat
 import sys
@@ -75,6 +76,37 @@ from tractor._testing import (
 _state: bool = False
 
 
+def _non_registration_contexts(
+    actor: Actor,
+) -> dict[tuple, str]:
+    return {
+        key: str(ctx._nsf)
+        for key, ctx in actor._contexts.items()
+        if str(ctx._nsf) != (
+            'tractor.discovery._registry:'
+            'Registrar.register_actor'
+        )
+    }
+
+
+@tractor.context
+async def startup_cancel_target(
+    ctx: Context,
+    started_path: str,
+    cancelled_path: str,
+) -> None:
+    Path(started_path).touch()
+    try:
+        await ctx.started()
+        await trio.sleep_forever()
+    finally:
+        Path(cancelled_path).touch()
+
+
+async def return_one() -> int:
+    return 1
+
+
 @tractor.context
 async def too_many_starteds(
     ctx: Context,
@@ -168,6 +200,92 @@ async def assert_state(value: bool):
     assert _state == value
 
 
+@tractor_test
+async def test_cancel_during_context_startup(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    Cancel a context after sending `Start` but before its ack.
+
+    `Portal.open_context()` allocates its caller-side `Context` while
+    entering the async context manager. Cancellation used to strand
+    that local context and leave the remote target running. The patched
+    `Channel.send()` publishes `Start`, then blocks before
+    `Actor.start_remote_task()` can await `StartAck`. Cancelling the
+    caller proves cleanup issues one bounded, non-recursive cancel RPC,
+    stops the target and removes both helper contexts. A subsequent
+    RPC proves the caller-owned actor remains usable.
+
+    '''
+    started_path = tmp_path / 'startup_started'
+    cancelled_path = tmp_path / 'startup_cancelled'
+    start_sent = trio.Event()
+    original_send = tractor.Channel.send
+
+    async def delay_after_start(
+        chan: tractor.Channel,
+        payload: object,
+        hide_tb: bool = False,
+    ) -> None:
+        await original_send(
+            chan,
+            payload,
+            hide_tb=hide_tb,
+        )
+        if isinstance(payload, tractor.msg.Start):
+            if payload.func == 'startup_cancel_target':
+                start_sent.set()
+            await trio.sleep_forever()
+
+    async def open_target(
+        portal: tractor.Portal,
+    ) -> None:
+        async with portal.open_context(
+            startup_cancel_target,
+            started_path=str(started_path),
+            cancelled_path=str(cancelled_path),
+        ):
+            raise AssertionError('context startup should be cancelled')
+
+    async with tractor.open_nursery() as an:
+        actor = tractor.current_actor()
+        portal: tractor.Portal = await an.start_actor(
+            'startup_cancel_worker',
+            enable_modules=[__name__],
+        )
+        contexts_before = _non_registration_contexts(actor)
+        monkeypatch.setattr(
+            tractor.Channel,
+            'send',
+            delay_after_start,
+        )
+
+        async with trio.open_nursery() as tn:
+            tn.start_soon(open_target, portal)
+            with trio.fail_after(5):
+                await start_sent.wait()
+                while not started_path.exists():
+                    await trio.sleep(0.01)
+            tn.cancel_scope.cancel()
+
+        monkeypatch.setattr(
+            tractor.Channel,
+            'send',
+            original_send,
+        )
+        assert cancelled_path.exists()
+        assert _non_registration_contexts(actor) == contexts_before
+        assert await portal.run_from_ns(
+            __name__,
+            'return_one',
+        ) == 1
+        assert _non_registration_contexts(actor) == contexts_before
+        await portal.cancel_actor()
+
+
 @pytest.mark.parametrize(
     'error_parent',
     [False, ValueError, KeyboardInterrupt],

tractor/_context.py

@@ -1108,10 +1108,11 @@ class Context:
                 # NOTE: we're telling the far end actor to cancel a task
                 # corresponding to *this actor*. The far end local channel
                 # instance is passed to `Actor._cancel_task()` implicitly.
-                await self._portal.run_from_ns(
+                await self._portal._run_from_ns(
                     'self',
                     '_cancel_task',
-                    cid=cid,
+                    kwargs={'cid': cid},
+                    cancel_on_startup=False,
                 )
 
             if cs.cancelled_caught:

tractor/runtime/_portal.py

@@ -379,6 +379,38 @@ class Portal:
 
             return False
 
+    async def _run_from_ns(
+        self,
+        namespace_path: str,
+        function_name: str,
+        kwargs: dict[str, Any],
+        cancel_on_startup: bool = True,
+    ) -> Any:
+        '''
+        Run a namespace target with local startup policy controls.
+
+        '''
+        nsf = NamespacePath(
+            f'{namespace_path}:{function_name}'
+        )
+        ctx: Context = await self.actor.start_remote_task(
+            chan=self.channel,
+            nsf=nsf,
+            kwargs=kwargs,
+            portal=self,
+            cancel_on_startup=cancel_on_startup,
+        )
+        try:
+            return await ctx._pld_rx.recv_pld(
+                ipc=ctx,
+                expect_msg=Return,
+            )
+        finally:
+            self.actor._drop_context(ctx)
+            if not ctx._rx_chan._closed:
+                with trio.CancelScope(shield=True):
+                    await ctx._rx_chan.aclose()
+
     # TODO: do we still need this for low level `Actor`-runtime
     # method calls or can we also remove it?
     async def run_from_ns(
@@ -404,18 +436,10 @@ class Portal:
 
         '''
         __runtimeframe__: int = 1  # noqa
-        nsf = NamespacePath(
+        return await self._run_from_ns(
-            f'{namespace_path}:{function_name}'
+            namespace_path,
-        )
+            function_name,
-        ctx: Context = await self.actor.start_remote_task(
-            chan=self.channel,
-            nsf=nsf,
             kwargs=kwargs,
-            portal=self,
-        )
-        return await ctx._pld_rx.recv_pld(
-            ipc=ctx,
-            expect_msg=Return,
         )
 
     # TODO: factor this out into a `.highlevel` API-wrapper that uses

tractor/runtime/_runtime.py

@@ -784,6 +784,7 @@ class Actor:
         allow_overruns: bool = False,
         load_nsf: bool = False,
         ack_timeout: float = float('inf'),
+        cancel_on_startup: bool = True,
 
     ) -> Context:
         '''
@@ -835,13 +836,42 @@ class Actor:
 
             f'{pretty_struct.pformat(msg)}'
         )
-        await chan.send(msg)
+        try:
+            await chan.send(msg)
 
-        # NOTE wait on first `StartAck` response msg and validate;
+            # NOTE wait on first `StartAck` response msg and validate;
-        # this should be immediate and does not (yet) wait for the
+            # this should be immediate and does not (yet) wait for the
-        # remote child task to sync via `Context.started()`.
+            # remote child task to sync via `Context.started()`.
-        with trio.fail_after(ack_timeout):
+            with trio.fail_after(ack_timeout):
-            first_msg: msgtypes.StartAck = await ctx._rx_chan.receive()
+                first_msg: msgtypes.StartAck = await ctx._rx_chan.receive()
+
+        except trio.Cancelled:
+            with trio.CancelScope(shield=True):
+                # `MsgpackTransport.send()` closes its stream when
+                # cancellation interrupts the length-prefixed write
+                # because an unknown prefix may already be sent. A
+                # connected channel means cancellation happened before
+                # that write or after it completed, so `_cancel_task`
+                # is protocol-safe (and a no-op if `Start` was unsent).
+                if (
+                    cancel_on_startup
+                    and
+                    chan.connected()
+                ):
+                    try:
+                        await ctx.cancel()
+                    except BaseException as cancel_err:
+                        log.warning(
+                            'Failed to cancel RPC task during '
+                            'startup?\n'
+                            f'{cancel_err!r}\n'
+                        )
+
+                self._drop_context(ctx)
+                if not ctx._rx_chan._closed:
+                    await ctx._rx_chan.aclose()
+
+            raise
         try:
             functype: str = first_msg.functype
         except AttributeError: