fix(session): one supersession predicate, asked the same way everywhere

Scoping the arm-duty gate left four sibling gates in the same streaming
turn still comparing generations with a bare !=, so one function could
reach opposite verdicts for one generation shape: a Stop finalized the
display and stashed the partial where a Ctrl-C on the identical shape
did neither. _generation_superseded() is now the single predicate and
every site asks it — the cancel ref, the streaming consumer, the
dead-partial promotion, the Ctrl-C arm, and the orphan arm.

Each caller still performs its own read. That is the point rather than
an accident: the consumer's read is a genuine second look after the
ref's, and a consumer that delegated to the ref would inherit its stale
answer and run the arm duties for an orphan — nulling the successor's
usage slots and recording health for an abandoned lane.

Tests: TestSupersessionVerdictAgreement pins that the arms agree, in
both directions. Its orphan case pins the stronger invariant it turned
out to hold — a superseded generation never reaches an arm at all,
because the ref reads superseded and model_turn refuses to dispatch. The
last two hand-rolled dataclasses in the suite are replaced by the real
ToolCallDelta, and the prepare_wire docstring paragraph is re-flowed.

tests/test_cancel.py

@@ -4,13 +4,12 @@ import contextlib
 import json
 import threading
 import time
-from dataclasses import dataclass
 from unittest.mock import MagicMock, patch
 
 import pytest
 
 from tests._session_helpers import arm_session, make_session
-from turnstone.core.providers import StreamChunk
+from turnstone.core.providers import StreamChunk, ToolCallDelta
 from turnstone.core.session import (
     GenerationCancelled,
     _CancelRef,
@@ -193,21 +192,13 @@ class TestCancelDuringToolExecution:
         ui = NullUI()
         session = _make_session(ui=ui)
 
-        @dataclass
-        class FakeToolDelta:
-            index: int = 0
-            id: str = ""
-            name: str = ""
-            arguments_delta: str = ""
-
         # First call: return content with a tool call
         def stream_with_tool():
             yield StreamChunk(
-                tool_call_deltas=[FakeToolDelta(index=0, id="tc_1", name="bash")],
-                finish_reason="",
+                tool_call_deltas=[ToolCallDelta(index=0, id="tc_1", name="bash")],
             )
             yield StreamChunk(
-                tool_call_deltas=[FakeToolDelta(index=0, arguments_delta='{"command":"echo hi"}')],
+                tool_call_deltas=[ToolCallDelta(index=0, arguments_delta='{"command":"echo hi"}')],
                 finish_reason="tool_calls",
             )
 
@@ -325,23 +316,16 @@ class TestStreamFlushBeforeToolCalls:
         ui = TrackingUI()
         session = _make_session(ui=ui)
 
-        @dataclass
-        class FakeToolDelta:
-            index: int = 0
-            id: str = ""
-            name: str = ""
-            arguments_delta: str = ""
-
         def stream_content_then_tool():
             # Content long enough to leave chars in the tag-scan carry
             # buffer (ThinkTagSplitter retains the last MAX_TAG_LEN = 12
             # chars until a flush)
             yield StreamChunk(content_delta="Hello world, this is a test message")
             yield StreamChunk(
-                tool_call_deltas=[FakeToolDelta(index=0, id="tc_1", name="bash")],
+                tool_call_deltas=[ToolCallDelta(index=0, id="tc_1", name="bash")],
             )
             yield StreamChunk(
-                tool_call_deltas=[FakeToolDelta(index=0, arguments_delta='{"command":"echo hi"}')],
+                tool_call_deltas=[ToolCallDelta(index=0, arguments_delta='{"command":"echo hi"}')],
                 finish_reason="tool_calls",
             )
 
@@ -1208,3 +1192,83 @@ class TestOrphanArmDutiesGate:
         assert session._last_usage is None
         assert session._assistant_pending_tokens == 0
         tracker.record_success.assert_called_once()
+
+
+class TestSupersessionVerdictAgreement:
+    """Every arm of one streaming turn must reach the SAME supersession
+    verdict for the same generation shape.  Generation 0 is unscoped, so
+    on a session whose generation was claimed earlier a Stop and a Ctrl-C
+    must both finalize the display — a per-arm spelling once split them,
+    finalizing on one path and not the other."""
+
+    def _session_at_generation(self, gen, ui):
+        session = _make_session(ui=ui)
+        session._generation = gen
+        session.messages.append(Turn.user("hi"))
+        return session
+
+    def _drive(self, gen, kind):
+        ui = NullUI()
+        session = self._session_at_generation(gen, ui)
+
+        def stream():
+            yield StreamChunk(content_delta="partial answer")
+            if kind == "stop":
+                session.cancel()
+                yield StreamChunk(content_delta=" unreachable")
+            else:
+                raise KeyboardInterrupt
+
+        provider = arm_session(session, stream())
+        assert provider is session._provider
+        raised = None
+        try:
+            session._stream_response(0)
+        except BaseException as exc:  # noqa: BLE001 — the class IS the observation
+            raised = type(exc).__name__
+        return raised, ui.stream_ends, session._cancelled_partial_msg
+
+    def test_stop_and_ctrl_c_agree_on_an_unscoped_generation(self, tmp_db):
+        stop_raised, stop_ends, partial = self._drive(3, "stop")
+        kb_raised, kb_ends, _ = self._drive(3, "ctrl_c")
+
+        assert stop_raised == "GenerationCancelled"
+        assert kb_raised == "KeyboardInterrupt"
+        # The verdict is "live" on BOTH arms: each finalizes the display.
+        assert stop_ends == 1
+        assert kb_ends == 1
+        assert partial and partial["content"] == "partial answer"
+
+    def test_superseded_generation_finalizes_on_neither_arm(self, tmp_db):
+        # The scoped counterpart: a real orphan (its generation lost the
+        # claim) must touch the UI on no arm at all.  It never reaches
+        # one: the ref reads superseded, so ``model_turn`` refuses to
+        # dispatch and the ladder converts that to a cancel — an orphan
+        # issues no request and finalizes nothing.
+        ui = NullUI()
+        session = self._session_at_generation(5, ui)
+
+        def stream():
+            raise KeyboardInterrupt
+            yield  # unreachable; makes this a generator
+
+        provider = arm_session(session, stream())
+        with pytest.raises(GenerationCancelled):
+            session._stream_response(2)  # generation 2 lost the claim to 5
+        assert ui.stream_ends == 0
+        provider.create_streaming.assert_not_called()
+
+    def test_unscoped_generation_finalizes_on_the_ctrl_c_arm(self, tmp_db):
+        # Same arm, unscoped generation: the verdict flips to "live", so
+        # the display IS finalized — the agreement this class pins.
+        ui = NullUI()
+        session = self._session_at_generation(5, ui)
+
+        def stream():
+            raise KeyboardInterrupt
+            yield
+
+        arm_session(session, stream())
+        with pytest.raises(KeyboardInterrupt):
+            session._stream_response(0)
+        assert ui.stream_ends == 1

turnstone/core/model_turn.py

@@ -965,16 +965,15 @@ def model_turn(
     reads the error at all, which is what keeps a Stop mid-summary off
     the red-error path.
 
-    *prepare_wire* is the caller's OWN deterministic lowering, called
-    with the serving lane so per-lane capability posture is available,
-    composed after the seam passes and before the Phase-5 attach: the
-    main loop's system-message prepend, sender labels,
-    capability-sensitive system-turn fold, empty-user drop, and orphan
-    repair live here, each
-    a ``lowering.py``-composed pass.  It must be pure lowering — no
-    learned selection, no provider calls, no side effects (the session's
-    debug request dump, a read-only latch, is the tolerated exception).
-    The list that went to the wire comes back as
+    *prepare_wire* is the caller's OWN deterministic lowering, called with
+    the serving lane so per-lane capability posture is available, composed
+    after the seam passes and before the Phase-5 attach: the main loop's
+    system-message prepend, sender labels, capability-sensitive
+    system-turn fold, empty-user drop, and orphan repair live here, each a
+    ``lowering.py``-composed pass.  It must be pure lowering — no learned
+    selection, no provider calls, no side effects (the session's debug
+    request dump, a read-only latch, is the tolerated exception).  The
+    list that went to the wire comes back as
     ``ModelTurnResult.wire_msgs``.
 
     *deferred_names* passes through to ``create_streaming``: the

turnstone/core/session.py

@@ -327,6 +327,20 @@ class _CompactionIrreducibleError(Exception):
     """
 
 
+def _generation_superseded(session: ChatSession, my_generation: int) -> bool:
+    """Whether a newer generation has claimed *session* — THE supersession
+    predicate, one spelling for every site that asks.
+
+    Generation 0 is UNSCOPED (a direct seam caller, no ``send()`` above
+    it): it is never superseded, matching :meth:`ChatSession._check_cancelled`.
+    Callers read through this rather than sharing a cached answer — each
+    read is its own, which is what makes the streaming consumer's read a
+    real second look after the cancel ref's (the two-step supersession
+    window a force-cancel can land inside).
+    """
+    return bool(my_generation and session._generation != my_generation)
+
+
 class _CancelRef(list[Any]):
     """List proxy handed to ``create_streaming`` as its ``cancel_ref``.
 
@@ -379,8 +393,7 @@ class _CancelRef(list[Any]):
         self._armed = False
 
     def _superseded(self) -> bool:
-        gen = self._my_generation
-        return bool(gen and self._session._generation != gen)
+        return _generation_superseded(self._session, self._my_generation)
 
     @property
     def armed(self) -> bool:
@@ -543,8 +556,7 @@ class _StreamTurnConsumer:
         way (generation 0 is unscoped, so a direct seam caller is never
         superseded).  The two must agree: the ref decides whether to fire
         the arm hook, and the hook's own gate decides whether to act."""
-        gen = self._my_generation
-        return bool(gen and self._session._generation != gen)
+        return _generation_superseded(self._session, self._my_generation)
 
     def on_stream_armed(self) -> None:
         """`_CancelRef.on_first_append` — the request-accepted instant.
@@ -8306,7 +8318,7 @@ class ChatSession:
             text the user actually saw.  Never writes for a superseded
             generation — an orphan must not touch the successor's slot.
             """
-            if self._generation != my_generation:
+            if _generation_superseded(self, my_generation):
                 return
             if (
                 self._cancelled_partial_msg is None
@@ -8351,7 +8363,7 @@ class ChatSession:
                 # but send() treats Ctrl-C as a survivable, recorded path,
                 # so the dead attempt still needs its client-side finalize
                 # (the CLI's markdown fence resets only in on_stream_end).
-                if self._generation == my_generation:
+                if not _generation_superseded(self, my_generation):
                     self.ui.on_stream_end()
                 raise
             except Exception as e:
@@ -8365,7 +8377,7 @@ class ChatSession:
                     # re-create window reads the DEAD attempt's armed
                     # state (see ``end_attempt``).
                     consumer.end_attempt()
-                if self._generation != my_generation:
+                if _generation_superseded(self, my_generation):
                     # Superseded (force-cancel started a newer generation):
                     # an orphaned thread must not touch the UI — a finalize
                     # emitted here would clobber the NEW generation's