fix(*): let the page's stop cancel a delegated result re-entering

raven/rpc/methods/turn.py

@@ -722,11 +722,12 @@ async def turn_unsubscribe(
 
 
 def _cancel_payload(turn_id: str) -> dict[str, Any]:
-    """The cancelled-turn error's payload, carrying its turn when one is bound.
+    """The cancelled-turn error's payload, carrying the cancelled turn's id when there is one.
 
-    The lane's bound turn is the right source here and only here: the client
-    cancels its own session, so the turn ``turn.send`` bound to this lane is the
-    turn being cancelled. The key is omitted rather than sent empty, so that
+    The id is whatever identified the turn being cancelled: the one a named
+    cancel gave, a pending inject's own, or the turn ``turn.send`` bound to
+    this lane -- never the lane's slot read for a turn it does not hold. The
+    key is omitted rather than sent empty, so that
     "absent" has one representation on the wire -- a consumer that correlates a
     request to a turn treats both the same way, as not its own.
     """
@@ -750,18 +751,32 @@ async def turn_cancel(
     """``turn.cancel`` — cancel the in-flight turn + notify subscribers.
 
     Sequence:
-      1. Look up the active turn handle; if absent → ``{cancelled: False}``.
-      2. ``await handle.cancel()``.
+      1. Find the turn to cancel. With ``turn_id``: the lane's running turn,
+         and only when it carries that id -- whether ``turn.send`` bound it or
+         the runtime opened it (a delegated result re-entering, which binds no
+         handle). Without: the active turn handle, else the oldest pending
+         inject. Neither there → ``{cancelled: False}``.
+      2. Cancel what was found.
       3. ``emitter.emit(session_key, error(reason="cancelled_by_client"))`` — the
          client resets its UI off this event. This is the ONLY cancelled-turn
          error; the sink stays silent on a cancelled TurnFailed (avoiding a
          double error), so this emit is the one signal that clears the front-end
          turn slot — it must always fire.
-      4. Await the handle so the turn is provably unwound (the sink's TurnFailed
+      4. Await the turn so it is provably unwound (the sink's TurnFailed
          handler drops the active-turn slot) before returning, so the next
          ``turn.send`` cannot race a half-unwound turn into a phantom -32003.
       5. Return ``{cancelled: True}``.
 
+    A named cancel goes past pending injects to the running turn: an inject
+    already merged into it ends with it, as it does when a bound turn is
+    cancelled, and one still waiting in the lane's mailbox falls back to a turn
+    of its own once the cancelled turn unwinds -- the spine never drops it, so
+    the reader's correction runs rather than vanishing under the stop. An
+    unnamed cancel keeps its older reach on purpose: callers that send no id
+    (the terminal UI, the ACP adapter's ``session/cancel``) mean the turn they
+    started, and reaching for whatever the lane happens to be running would
+    stop a cron wake, or a delegated turn that began after the request left.
+
     The subscription is SESSION-scoped, not turn-scoped: a per-turn cancel ends
     only the turn and MUST leave the session's subscriptions open so the next
     turn's events still reach the client.
@@ -780,6 +795,26 @@ async def turn_cancel(
         else parsed.session_key
     )
 
+    if parsed.turn_id is not None:
+        # Probed, not assumed, like ``_lane_in_flight``: a scheduler without
+        # the seam (a test fake) has no running turn to name.
+        probe = getattr(_scheduler, "running_turn", None)
+        running = probe(lane) if callable(probe) else None
+        if running is None or running[0] != parsed.turn_id:
+            return {"cancelled": False}
+        handle = running[1]
+        await handle.cancel()
+        if emitter is not None:
+            await emitter.emit(
+                parsed.session_key,
+                {
+                    "type": "error",
+                    "payload": _tag(_cancel_payload(parsed.turn_id), (direct_targets or {}).get(lane)),
+                },
+            )
+        await handle.result()
+        return {"cancelled": True}
+
     reach = _cancellable(lane)
     if reach is None:
         return {"cancelled": False}

raven/rpc/models.py

@@ -1745,6 +1745,12 @@ class TurnCancelParams(_Strict):
     # handler looks up a key nothing was ever registered under and reports
     # "nothing to cancel" for a turn that is plainly running.
     target: DirectTarget | None = None
+    # The turn to cancel, by the id its opening frame carried. Present, the
+    # cancel reaches the lane's running turn only when that turn has this id,
+    # whoever opened it -- the one way to stop a turn the runtime opened (a
+    # delegated result re-entering), which binds no handle. Absent, the cancel
+    # reaches only what this surface bound: a turn.send turn or a pending inject.
+    turn_id: str | None = None
 
 
 class TurnCancelResult(_Strict):

raven/spine/scheduler.py

@@ -166,6 +166,10 @@ def __init__(self, runner: TurnRunner, pools: OriginPools, sink: EventSink, conv
         self._worker: asyncio.Task | None = None
         self._run_task: asyncio.Task | None = None
         self._running_fut: asyncio.Future | None = None
+        # The running turn's id, set and cleared with ``_running_fut``: a caller
+        # that names a turn (turn.cancel's ``turn_id``) has nothing else to
+        # match against, since the request itself lives only in the task.
+        self._running_turn_id: str | None = None
         self._idle_since: float | None = None  # set when the worker drains; the reaper's clock
         # Whether the running turn's payload has already emitted its own terminal
         # event. Deliberately not "did the payload start": a cancel can land while
@@ -293,6 +297,15 @@ def running_future(self) -> asyncio.Future | None:
         """
         return self._running_fut
 
+    def running_turn(self) -> tuple[str, asyncio.Future] | None:
+        """The running turn's id and terminal future while its task can still be
+        cancelled, else None -- idle, or between the task ending and the worker
+        resolving it, when a cancel could only arrive after the answer.
+        """
+        if self._running_fut is None or self._run_task is None or self._run_task.done():
+            return None
+        return self._running_turn_id or "", self._running_fut
+
     def drain_pending(self) -> int:
         """Resolve every not-yet-running turn (queued + mailboxed injects) as
         cancelled and return the count, leaving the running turn untouched.
@@ -336,6 +349,7 @@ async def _run_worker(self) -> None:
         while self._pending:
             req, fut = self._pending.popleft()
             self._running_fut = fut
+            self._running_turn_id = req.turn_id
             # Create the task synchronously (no await before this) so the turn is
             # cancel-visible via _run_task the moment it leaves the queue. Which is
             # also why the flag is armed here: from this line until the payload's
@@ -416,6 +430,7 @@ async def _run_worker(self) -> None:
                     self._enqueue(inject_req, inject_fut)
                 self._run_task = None
                 self._running_fut = None
+                self._running_turn_id = None
         # Idle exit: stamp the reaper's silence clock, with no await between the
         # queue check and return — a submit racing the exit must not be lost
         # (lost-wakeup), and the same gap keeps the stamp atomic against submit
@@ -561,8 +576,14 @@ async def result(self) -> TurnOutcome | TurnFailed | None:
         """How the turn ended, in three: the ``TurnOutcome`` of a turn that ran to
         an answer, the ``TurnFailed`` its lane filed for a turn that failed with a
         name, or None when it was cancelled or never got to run.
+
+        Shielded: the lane's worker is this future's sole resolver, and a waiter
+        cancelled mid-await -- a connection closing under the handler waiting
+        here -- would otherwise cancel the future itself, so the worker's own
+        ``set_result`` raises and it dies with the lane still marked running.
+        The waiter is cancelled; the turn and its future are not.
         """
-        return await self._fut
+        return await asyncio.shield(self._fut)
 
     async def cancel(self) -> None:
         await self._lane.cancel_turn(self._fut)
@@ -611,6 +632,22 @@ def cancel_conversation(self, conversation_id: str) -> int:
         lane = self._lanes.get(conversation_id)
         return lane.cancel() if lane is not None else 0
 
+    def running_turn(self, conversation_id: str) -> tuple[str, TurnHandle] | None:
+        """The turn running on a conversation's lane, as its id and a handle,
+        whoever submitted it; None when nothing there can still be cancelled.
+
+        A handle rather than the lane's future, so a caller reaches the turn
+        only through what every submitter gets: ``cancel`` ends the running
+        turn and leaves queued work in place, and ``result`` waits without
+        being able to cancel the future the worker resolves.
+        """
+        lane = self._lanes.get(conversation_id)
+        running = lane.running_turn() if lane is not None else None
+        if running is None:
+            return None
+        turn_id, fut = running
+        return turn_id, TurnHandle(lane, fut)
+
     def has_inflight(self, conversation_id: str) -> bool:
         """True if a turn is currently running for this conversation's lane.
         The inbound gate uses this to submit a mid-turn message as

rpc-schema/openrpc.json

@@ -805,6 +805,14 @@
           "schema": {
             "$ref": "#/components/schemas/DirectTarget"
           }
+        },
+        {
+          "name": "turn_id",
+          "description": "The turn to cancel, by the id its opening frame carried. Present, the lane's running turn is cancelled only when it has this id, whoever opened it -- the way to stop a turn the runtime opened, which binds no handle. Absent, only a turn.send turn or a pending inject is reached.",
+          "required": false,
+          "schema": {
+            "type": "string"
+          }
         }
       ],
       "result": {