Clean failed remote-task startup state

`Actor.start_remote_task()` registers its caller context before
sending `Start`, but only cancellation cleaned that state.
Encoding, ack timeout, malformed ack and remote authorization
errors leaked it.

Protect the complete send, acknowledgement and validation phase.
Track successful publication, make a remote cancellation attempt
only when protocol-safe and always release the local context while
preserving the original startup error.

Cover both pre-publication serialization failure and a remote
`ModuleNotExposed` rejection without damaging a reused portal.

Caught-during: review remediation
Found-via: `/run-tests` startup-failure regressions

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

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

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

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

@@ -0,0 +1,36 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: ses_3e4c90d3eafeqHEtRYSIHgHhpA
+timestamp: 2026-08-18T19:30:04Z
+git_ref: bf06b4f8
+scope: code
+substantive: true
+raw_file: 20260818T193004Z_bf06b4f8_prompt_io.raw.md
+---
+
+## Prompt
+
+Release caller-side context state for every remote-task startup failure,
+not only local cancellation. Preserve the remote error, avoid unsafe
+follow-up sends and prove pre-publication serialization failures leave
+a reused portal healthy.
+
+## Response summary
+
+Extend remote-task startup cleanup across send, acknowledgement and
+validation errors. Track completed publication, perform only safe
+best-effort cancellation and deterministically remove local state.
+
+## Files changed
+
+- `tractor/runtime/_runtime.py` - clean every startup failure path.
+- `tests/test_context_stream_semantics.py` - cover authorization and
+  serialization failures before context entry.
+
+## Human edits
+
+The human accepted the discovered edge-case fixes but required general
+startup cleanup to land separately from cancellation cleanup, transport
+integrity and the public API. This boundary preserves that behavioral
+distinction and its dedicated commit-message rationale.

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

@@ -0,0 +1,19 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-18T19:30:04Z
+git_ref: bf06b4f8
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+`Actor.start_remote_task()` inserts a context before sending `Start`,
+but startup errors other than cancellation escape without removing or
+closing that caller state. Serialization errors, acknowledgement
+timeouts, malformed acknowledgements and remote authorization errors
+can therefore leak context-registry entries.
+
+> `git diff HEAD~1..HEAD -- tractor/runtime/_runtime.py tests/test_context_stream_semantics.py`
+
+Cover the complete send, acknowledgement and validation phase with
+exceptional cleanup. Attempt remote cancellation only when publication
+is known complete or protocol-safe, and always release local state.

tests/test_context_stream_semantics.py

@@ -286,6 +286,82 @@ async def test_cancel_during_context_startup(
         await portal.cancel_actor()
 
 
+@tractor_test
+async def test_start_serialization_error_cleans_context(
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    Deallocate caller state when `Start` can not be serialized.
+
+    `Actor.start_remote_task()` registers its caller-side `Context`
+    before encoding the request. An unsupported argument used to raise
+    `MsgTypeError` before publication while leaking that registry
+    entry. Comparing the context registry around the failed start
+    proves cleanup, and a following valid context proves no bytes
+    reached or damaged the reused portal's transport.
+
+    '''
+    async with tractor.open_nursery() as an:
+        actor = tractor.current_actor()
+        portal: tractor.Portal = await an.start_actor(
+            'serialization_error_worker',
+            enable_modules=[__name__],
+        )
+        contexts_before = _non_registration_contexts(actor)
+        with pytest.raises(tractor.MsgTypeError):
+            async with portal.open_context(
+                simple_setup_teardown,
+                data=object(),
+            ):
+                raise AssertionError('invalid `Start` was accepted')
+
+        assert _non_registration_contexts(actor) == contexts_before
+        async with portal.open_context(
+            simple_setup_teardown,
+            data=1,
+        ) as (ctx, started):
+            assert started == 2
+            assert await ctx.wait_for_result() == 'yo'
+
+        assert _non_registration_contexts(actor) == contexts_before
+        await portal.cancel_actor()
+
+
+@tractor_test
+async def test_start_module_error_cleans_context(
+    start_method: str,
+    debug_mode: bool,
+):
+    '''
+    Deallocate caller state after a remote startup rejection.
+
+    A target actor without this test module rejects the requested
+    context before sending `StartAck`. That remote
+    `ModuleNotExposed` used to escape startup validation while leaving
+    the caller context registered. The boxed error and before/after
+    registry comparison prove the remote failure remains visible and
+    local startup state is released.
+
+    '''
+    async with tractor.open_nursery() as an:
+        actor = tractor.current_actor()
+        portal: tractor.Portal = await an.start_actor(
+            'module_error_worker',
+        )
+        contexts_before = _non_registration_contexts(actor)
+        with pytest.raises(tractor.RemoteActorError) as excinfo:
+            async with portal.open_context(
+                simple_setup_teardown,
+                data=1,
+            ):
+                raise AssertionError('unexposed context was started')
+
+        assert excinfo.value.boxed_type is tractor.ModuleNotExposed
+        assert _non_registration_contexts(actor) == contexts_before
+        await portal.cancel_actor()
+
+
 @pytest.mark.parametrize(
     'error_parent',
     [False, ValueError, KeyboardInterrupt],

tractor/runtime/_runtime.py

@@ -836,8 +836,10 @@ class Actor:
 
             f'{pretty_struct.pformat(msg)}'
         )
+        start_published: bool = False
         try:
             await chan.send(msg)
+            start_published = True
 
             # NOTE wait on first `StartAck` response msg and validate;
             # this should be immediate and does not (yet) wait for the
@@ -845,7 +847,22 @@ class Actor:
             with trio.fail_after(ack_timeout):
                 first_msg: msgtypes.StartAck = await ctx._rx_chan.receive()
 
-        except trio.Cancelled:
+            try:
+                functype: str = first_msg.functype
+            except AttributeError:
+                raise unpack_error(first_msg, chan)
+
+            if functype not in (
+                'asyncfunc',
+                'asyncgen',
+                'context',
+            ):
+                raise ValueError(
+                    f'Invalid `StartAck.functype: str = '
+                    f'{first_msg!r}` ??'
+                )
+
+        except BaseException as startup_err:
             with trio.CancelScope(shield=True):
                 # `MsgpackTransport.send()` closes its stream when
                 # cancellation interrupts the length-prefixed write
@@ -856,7 +873,14 @@ class Actor:
                 if (
                     cancel_on_startup
                     and
+                    (
+                        start_published
+                        or (
+                            isinstance(startup_err, trio.Cancelled)
+                            and
                             chan.connected()
+                        )
+                    )
                 ):
                     try:
                         await ctx.cancel()
@@ -872,19 +896,6 @@ class Actor:
                     await ctx._rx_chan.aclose()
 
             raise
-        try:
-            functype: str = first_msg.functype
-        except AttributeError:
-            raise unpack_error(first_msg, chan)
-
-        if functype not in (
-            'asyncfunc',
-            'asyncgen',
-            'context',
-        ):
-            raise ValueError(
-                f'Invalid `StartAck.functype: str = {first_msg!r}` ??'
-            )
 
         ctx._remote_func_type = functype
         return ctx