diff --git a/ai/prompt-io/opencode/20260818T193003Z_bf06b4f8_prompt_io.md b/ai/prompt-io/opencode/20260818T193003Z_bf06b4f8_prompt_io.md new file mode 100644 index 00000000..88c1b722 --- /dev/null +++ b/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. diff --git a/ai/prompt-io/opencode/20260818T193003Z_bf06b4f8_prompt_io.raw.md b/ai/prompt-io/opencode/20260818T193003Z_bf06b4f8_prompt_io.raw.md new file mode 100644 index 00000000..03b94893 --- /dev/null +++ b/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. diff --git a/tests/test_context_stream_semantics.py b/tests/test_context_stream_semantics.py index 8ef85426..4e1572a3 100644 --- a/tests/test_context_stream_semantics.py +++ b/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], diff --git a/tractor/_context.py b/tractor/_context.py index 089442ef..8fd69743 100644 --- a/tractor/_context.py +++ b/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: diff --git a/tractor/runtime/_portal.py b/tractor/runtime/_portal.py index 738599ac..7c94d74c 100644 --- a/tractor/runtime/_portal.py +++ b/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( - f'{namespace_path}:{function_name}' - ) - ctx: Context = await self.actor.start_remote_task( - chan=self.channel, - nsf=nsf, + return await self._run_from_ns( + namespace_path, + function_name, 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 diff --git a/tractor/runtime/_runtime.py b/tractor/runtime/_runtime.py index a1788336..0c2d1c10 100644 --- a/tractor/runtime/_runtime.py +++ b/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; - # this should be immediate and does not (yet) wait for the - # remote child task to sync via `Context.started()`. - with trio.fail_after(ack_timeout): - first_msg: msgtypes.StartAck = await ctx._rx_chan.receive() + # NOTE wait on first `StartAck` response msg and validate; + # this should be immediate and does not (yet) wait for the + # remote child task to sync via `Context.started()`. + with trio.fail_after(ack_timeout): + 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: