Share `Context.cancel()` deadline with frame sends

A parent-side ctx cancel timeout previously bounded only the
remote `_cancel_task` ack. Complete-frame `send_all()` shielding
could hold request publication forever when a peer stopped reading.

Compute one absolute deadline and pass it through `._run_from_ns()`
so transport publication and the ack wait consume the same timeout
budget. Add a mock-clock regression which stalls the private RPC
under a nested shield and proves the transaction returns on time.

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

Prompt-IO: ai/prompt-io/opencode/20260824T222033Z_ce38cb6f_prompt_io.md

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

ai/prompt-io/opencode/20260824T222033Z_ce38cb6f_prompt_io.md

@@ -0,0 +1,37 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+session: d9d7df2c-7044-463f-8768-ec024718eac9
+timestamp: 2026-08-24T22:20:33Z
+git_ref: ce38cb6f
+scope: code
+substantive: true
+raw_file: 20260824T222033Z_ce38cb6f_prompt_io.raw.md
+---
+
+## Prompt
+
+Continue the PR #481 review after triage. The human explicitly accepted
+the proposed merge-blocking `Context.cancel()` deadline update with
+"keep" and required the work to remain limited to that review item.
+
+## Response summary
+
+Update `Context.cancel()` so one absolute deadline bounds both shielded
+cancel-request publication and acknowledgement waiting. Add a focused
+mocked-clock regression for the blocked-publication failure mode and run
+the narrow cancellation tests.
+
+## Files changed
+
+- `tractor/_context.py` - forward the cancel transaction's absolute
+  deadline to frame publication.
+- `tests/test_to_actor.py` - prove blocked context-cancel publication is
+  bounded by the shared deadline.
+
+## Human edits
+
+The human retained ownership of review scope and explicitly selected
+"keep" for this item after receiving keep/defer/drop options. The human
+required no unrelated cancellation changes and did not directly edit
+source lines.

ai/prompt-io/opencode/20260824T222033Z_ce38cb6f_prompt_io.raw.md

@@ -0,0 +1,26 @@
+---
+model: openai/gpt-5.6-sol
+service: opencode
+timestamp: 2026-08-24T22:20:33Z
+git_ref: ce38cb6f
+diff_cmd: git diff HEAD~1..HEAD
+---
+
+Implement the approved PR #481 review update for `Context.cancel()`.
+Use one absolute deadline for both cancellation-request frame
+publication and acknowledgement waiting, without broadening the change
+to unrelated cancellation behavior.
+
+> `git diff HEAD~1..HEAD -- tractor/_context.py`
+
+`Context.cancel()` computes one absolute cancellation deadline, uses it
+for the outer bounded wait, and forwards it through
+`Portal._run_from_ns()` to shielded frame publication.
+
+> `git diff HEAD~1..HEAD -- tests/test_to_actor.py`
+
+A deterministic mocked-clock regression arranges a shielded blocked
+publication and proves that `Context.cancel()` forwards the same deadline
+which bounds the complete cancel transaction.
+
+Run the focused cancellation deadline regressions after the edit.

tests/test_to_actor.py

@@ -319,6 +319,67 @@ def test_cancel_actor_timeout_closes_blocked_send():
     )
 
 
+def test_context_cancel_timeout_closes_blocked_send():
+    '''
+    Bound context-cancel publication and acknowledgement together.
+
+    `Context.cancel()` shields its transaction from outer cancellation,
+    while `MsgpackTransport.send()` separately shields complete frame
+    publication. Previously the context's timeout was not forwarded to
+    that inner shield, so a peer which stopped reading could leave the
+    cancel task blocked forever instead of respecting `timeout`.
+
+    The fake private RPC records the forwarded absolute deadline and
+    blocks under a send-like shield until that deadline. The mock clock
+    advances directly to it; completion and the exact recorded value
+    prove that publication shares the context's one-second budget.
+
+    '''
+    async def main() -> None:
+        deadlines: list[float] = []
+
+        async def blocked_cancel(
+            namespace: str,
+            function: str,
+            kwargs: dict[str, object],
+            cancel_on_startup: bool,
+            send_deadline: float,
+        ) -> None:
+            assert (namespace, function) == ('self', '_cancel_task')
+            assert kwargs == {'cid': 'blocked-context'}
+            assert not cancel_on_startup
+            deadlines.append(send_deadline)
+            with trio.CancelScope(
+                deadline=send_deadline,
+                shield=True,
+            ):
+                await trio.sleep_forever()
+            await trio.lowlevel.checkpoint_if_cancelled()
+
+        peer_aid = tractor.msg.Aid(
+            name='blocked_peer',
+            uuid='test',
+        )
+        ctx = object.__new__(tractor.Context)
+        ctx.chan = SimpleNamespace(
+            aid=peer_aid,
+            connected=lambda: True,
+            transport=SimpleNamespace(maddr='test://blocked'),
+        )
+        ctx.cid = 'blocked-context'
+        ctx._portal = SimpleNamespace(_run_from_ns=blocked_cancel)
+        ctx._nsf = NamespacePath.from_ref(add_one)
+
+        await ctx.cancel(timeout=1)
+
+        assert deadlines == [1.]
+
+    trio.run(
+        main,
+        clock=MockClock(autojump_threshold=0),
+    )
+
+
 def _mock_actor_nursery() -> tractor.ActorNursery:
     an = object.__new__(tractor.ActorNursery)
     an._children = {}

tractor/_context.py

@@ -1097,7 +1097,12 @@ class Context:
                 )
 
             cid: str = self.cid
-            with trio.move_on_after(timeout) as cs:
+            cancel_deadline: float = (
+                trio.current_time()
+                +
+                timeout
+            )
+            with trio.move_on_at(cancel_deadline) as cs:
                 cs.shield = True
                 log.cancel(
                     header
@@ -1113,6 +1118,7 @@ class Context:
                     '_cancel_task',
                     kwargs={'cid': cid},
                     cancel_on_startup=False,
+                    send_deadline=cancel_deadline,
                 )
 
             if cs.cancelled_caught: