diff --git a/tests/test_cancel.py b/tests/test_cancel.py index c4ade926..f326e2a5 100644 --- a/tests/test_cancel.py +++ b/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 diff --git a/turnstone/core/model_turn.py b/turnstone/core/model_turn.py index f5364a08..cc06d065 100644 --- a/turnstone/core/model_turn.py +++ b/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 diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 3e353778..7e46f735 100644 --- a/turnstone/core/session.py +++ b/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