diff --git a/CHANGELOG.md b/CHANGELOG.md index e3a6f92e..8e1648b2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -206,6 +206,30 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. ### Fixed +- **A cancelled judge, guard, or compaction call can now stop before its + request goes out (#972).** Previously it could not: `model_turn` refused + to *re-issue* an abandoned call after a mid-stream death, but nothing + checked before a first dispatch, so a call whose caller had already gone + away still sent — and the reply was discarded unread after the endpoint + had accepted the work. It now checks immediately before sending, so a + Stop observed by that point costs no request, and again on entry, so a + call already cancelled when it arrives also skips credential resolution. + Cancellation is cooperative, which bounds what that buys: a Stop only + saves the request if it lands before dispatch — sending is a moment, the + response streaming back is the rest of the call, and an abort arriving + then still meets a request in flight, closed in place exactly as before. + The window that did widen usefully is a delegated-auth alias whose token + mint blocks; a Stop during that mint now costs no request (though a mint + already under way still completes). What a stopped call saves is the + request, its prompt-side billing, and — on a capacity-bounded + self-hosted endpoint — a slot a live request wanted. Unchanged: the + interactive turn, which has its own pre-send cancellation check on a + different path, and the lanes that thread no cancellation handle + (attachment perception, title generation, web-fetch extraction, + sub-agents, optimizer, eval) — and web-fetch extraction deliberately + never will, since it runs on parallel tool threads where registering one + would clobber the main stream's. + - **Inline ``/`` blocks no longer leak into drained results (#965, #940).** On servers without a reasoning parser (parserless vLLM/llama.cpp, LM Studio, bare gateways), reasoning @@ -220,6 +244,7 @@ Earlier stable lines (`stable/1.6`, `stable/1.5`) are frozen. mismatched-vocabulary close tag (`…`) now closes the block — matching the interactive lane's long-standing rule — where the old per-lane strips treated it as unterminated. + - **A transport failure mid-generation no longer kills the interactive turn (#937).** A wire death during body streaming (TLS record failure, connection reset — `httpx.ReadError` and kin) surfaces after the diff --git a/tests/test_model_turn.py b/tests/test_model_turn.py index a36ae779..a0bed2d3 100644 --- a/tests/test_model_turn.py +++ b/tests/test_model_turn.py @@ -8,6 +8,7 @@ single-shot lanes (phase 2) can build on it without re-deriving semantics. from __future__ import annotations +import logging from types import SimpleNamespace from typing import Any from unittest.mock import MagicMock @@ -231,10 +232,17 @@ def test_model_turn_abort_during_backoff_suppresses_reissue( assert len(provider.calls) == 1 -def test_model_turn_does_not_retry_after_abort() -> None: - # A deadline that closed the stream must not have the request - # resurrected behind its back: the closed stream dies with an error - # that LOOKS retryable, but the aborted cancel_ref gates the re-issue. +def test_model_turn_does_not_retry_after_abort( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + # A call the deadline abandoned must not have its request resurrected + # behind its back: in the field the abort closes the stream and the + # drain dies with an error that LOOKS retryable. The fixture models + # only the shape of that — abort landing after dispatch, drain raising + # IncompleteStreamError — because the aborted ref is what gates the + # re-issue regardless of which of the two produced the error. This is + # the RE-ISSUE gate; of the tests below, two cover the pre-dispatch reads + # and the third pins the raised message. from turnstone.core.deadline import StreamAbortRef provider = _FlakyProvider( @@ -242,12 +250,107 @@ def test_model_turn_does_not_retry_after_abort() -> None: ) lane = ModelLane(provider=provider, client=object(), model="m") ref = StreamAbortRef() - ref.abort() + dispatch = provider.create_streaming - with pytest.raises(IncompleteStreamError): + def _abort_after_dispatch(**kwargs: Any) -> Any: + stream = dispatch(**kwargs) + ref.abort() # the deadline daemon fires; the request is already out + return stream + + monkeypatch.setattr(provider, "create_streaming", _abort_after_dispatch) + + with ( + caplog.at_level(logging.WARNING, logger="turnstone.core.model_turn"), + pytest.raises(IncompleteStreamError), + ): model_turn(lane, [Turn.user("x")], cancel_ref=ref) assert len(provider.calls) == 1 + # Isolates THIS gate from the post-backoff one its sibling covers. The + # abort is read where the failure surfaces, so the loop never announces a + # re-issue it will not make; delete that gate and the backoff arm still + # ends at one dispatch, but it logs on the way — which is what makes this + # assertion, and not the call count, the discriminating one. + assert "model_turn.drain_retry" not in caplog.text + + +def test_abort_landing_during_the_backend_auth_mint_still_never_dispatches() -> None: + # The window an entry-only check cannot see: on a dynamic-auth alias + # the resolve can block for seconds on a cache miss, so an abort can + # land after the entry read and before the request. The read + # immediately before create_streaming is what covers it. + from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef + + provider = _FakeProvider([CompletionResult(content="never")]) + ref = StreamAbortRef() + client = MagicMock() + + def _abort_during_mint(alias: str) -> str: + ref.abort() # the user hits Stop while the mint is blocked + return "minted-token" + + lane = ModelLane( + provider=provider, + client=client, + model="m", + alias="obo-gateway", + backend_auth_resolver=_abort_during_mint, + ) + + with pytest.raises(DeadlineCancelledError): + model_turn(lane, [Turn.user("x")], cancel_ref=ref) + + assert provider.calls == [] + + +def test_pre_dispatch_abort_precedes_the_backend_auth_mint() -> None: + # Placement of the FIRST read: an already-abandoned call skips the + # resolve entirely. On a cache miss that resolve is a network mint + # under a cluster-wide lock, so this is work worth not doing — but the + # invariant itself rides the read before create_streaming, not this one. + from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef + + provider = _FakeProvider([CompletionResult(content="never")]) + resolver = MagicMock(return_value="minted-token") + client = MagicMock() + lane = ModelLane( + provider=provider, + client=client, + model="m", + alias="obo-gateway", + backend_auth_resolver=resolver, + ) + ref = StreamAbortRef() + ref.abort() + + with pytest.raises(DeadlineCancelledError): + model_turn(lane, [Turn.user("x")], cancel_ref=ref) + + resolver.assert_not_called() + client.with_options.assert_not_called() + assert provider.calls == [] + + +def test_pre_dispatch_abort_does_not_read_as_a_context_overflow() -> None: + # A latent coupling, pinned deliberately rather than a live path: today + # compaction's ``except`` arm re-checks the session first and raises + # GenerationCancelled, and ``_stop_retrying`` short-circuits on the class + # gate, so this message never reaches ``_is_ctx_overflow``. It would the + # moment either shortcut moves — and ``_is_ctx_overflow`` classifies an + # unrecognized class by TEXT, so an overflow reading would send the + # compaction lane subdividing. The message is a wire contract; pin it. + from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef + from turnstone.core.session import _is_ctx_overflow + + provider = _FakeProvider([CompletionResult(content="never")]) + lane = ModelLane(provider=provider, client=object(), model="m") + ref = StreamAbortRef() + ref.abort() + + with pytest.raises(DeadlineCancelledError) as excinfo: + model_turn(lane, [Turn.user("x")], cancel_ref=ref) + + assert not _is_ctx_overflow(excinfo.value) def _real_semantics_store(**stored: Any) -> SimpleNamespace: diff --git a/turnstone/core/deadline.py b/turnstone/core/deadline.py index ce2a440b..476240f8 100644 --- a/turnstone/core/deadline.py +++ b/turnstone/core/deadline.py @@ -80,10 +80,16 @@ class StreamAbortRef(list[Any]): def aborted(self) -> bool: """Whether :meth:`abort` has fired. - ``model_turn``'s drain-retry gate reads this (duck-typed off any - ``cancel_ref``): an aborted stream dies with a transport error - that looks retryable, and re-issuing the request would resurrect - a call its deadline already abandoned. + ``model_turn`` reads this (duck-typed off any ``cancel_ref``) in + three roles, all load-bearing. On entry, so an abandoned call + skips the lowering and the credential resolve. Immediately + before it dispatches, so an abort observed by then costs no + request. And at its drain-retry gates, where an aborted stream + dies with a transport error that looks retryable and re-issuing + would resurrect a call its deadline already abandoned. No read + closes the window — an abort firing after the last one still + meets the arriving handle at :meth:`append`, which is why that + hook is not redundant with them. """ return self._aborted diff --git a/turnstone/core/model_turn.py b/turnstone/core/model_turn.py index 396fa478..a24993b7 100644 --- a/turnstone/core/model_turn.py +++ b/turnstone/core/model_turn.py @@ -15,7 +15,12 @@ Contract, held deliberately narrow: * **Policy-free.** No retry, no deadline, no tool execution, no usage recording inside — those belong to each caller. The callers are different organs (a judge is not a sub-agent is not a title generator); - the plant call is the one thing they share. + the plant call is the one thing they share. Two carve-outs, both about + a call that is already dead rather than about policy: the drain-retry + loop re-issues a mid-stream death, and an aborted ``cancel_ref`` raises + ``DeadlineCancelledError`` rather than dispatch — the caller still owns + the deadline, this only refuses to spend on a call already abandoned + (see ``Raises`` on :func:`model_turn`). * **Providers stay codegen.** The provider boundary keeps taking lowered wire dicts; Turn IR does not enter the provider Protocol, and ``lowering.py`` remains the only wire-mutation owner. This module @@ -48,6 +53,7 @@ if TYPE_CHECKING: UsageInfo, ) +from turnstone.core.deadline import DeadlineCancelledError from turnstone.core.history_decoration import attach_vllm_chat_reasoning_field from turnstone.core.log import get_logger from turnstone.core.lowering import ( @@ -633,6 +639,29 @@ def cap_tool_calls(result: ModelTurnResult, max_calls: int) -> tuple[list[dict[s return capped, turn +def _raise_if_aborted(cancel_ref: Any, lane: ModelLane) -> None: + """Refuse to go on with a call whose caller has already gone away (#972). + + Duck-typed on the same ``aborted`` predicate the drain-retry gate + reads, so a ``None`` ref — most lanes — and a plain-list ref stay + legal. One definition, two call sites in :func:`model_turn`: entry, + and immediately before ``create_streaming``. Nothing interrupts the + credential resolve that runs between them — a mint already under way + completes even when the abort lands inside it. The second read is + what turns such an abort into a skipped request rather than a sent + one, which is why it is not redundant with the first. + + The raised message is control flow, not prose. ``_is_ctx_overflow`` + classifies an exception class it does not recognize by TEXT, so + context-window vocabulary here would read downstream as a real + overflow and send the compaction lane subdividing. + """ + if not getattr(cancel_ref, "aborted", False): + return + log.debug("model_turn.abort_before_dispatch", model=lane.model, alias=lane.alias) + raise DeadlineCancelledError("cancel_ref aborted before dispatch") + + def model_turn( lane: ModelLane, turns: Sequence[Turn], @@ -695,7 +724,30 @@ def model_turn( stream object (which has ``.close()``) before the first chunk, so a deadline daemon can abort the blocked HTTP read from another thread instead of abandoning it (transport is a drained ``create_streaming`` - — see :func:`drain_stream`). + — see :func:`drain_stream`). Its ``aborted`` predicate gates every + dispatch, not only a re-issue (:func:`_raise_if_aborted`): an + abandoned call is not worth a request, and on a delegated-auth alias + not worth the credential resolve either — hence one read at entry, + ahead of lowering and ``lane.backend_auth_resolver``, and one + immediately before ``create_streaming``, which is the read that + actually keeps bytes off the wire. The entry read costs a pending + credential failure, which a pending abort now masks; right, because + the caller is gone. Neither read closes anything: a resolve already + under way finishes despite an abort landing inside it, and an abort + landing after the second read reaches the in-flight call the way it + always did — ``cancel_ref.append`` closes a handle that has not + arrived yet, ``abort`` closes one that has. + + The raise is :class:`~turnstone.core.deadline.DeadlineCancelledError`, + the deadline module's abandonment vocabulary. ``GenerationCancelled`` + is deliberately not raised here: it subclasses ``BaseException``, so + the ``except Exception`` arms around these calls cannot see it, and + it lives in ``session``, which imports this module. A caller whose + ref means something narrower than "this call is abandoned" owns the + translation — compaction does exactly that, its ``except`` arm + re-checking the session and raising ``GenerationCancelled`` before it + reads the error at all, which is what keeps a Stop mid-summary off + the red-error path. Raises whatever the provider raises — retry/deadline/fallback policy is the caller's — EXCEPT transient mid-stream deaths: a failure the @@ -707,7 +759,10 @@ def model_turn( that retry's new home (request-time failures still get the SDK's own policy inside ``create_streaming`` and propagate unchanged). An aborted *cancel_ref* suppresses retries — a deadline that closed the - stream must not have the request resurrected behind its back. + stream must not have the request resurrected behind its back — and, + read before each dispatch, raises + :class:`~turnstone.core.deadline.DeadlineCancelledError` instead of + issuing the request at all (see *cancel_ref* above). *backend_auth_token* is a delegated-user or app-identity credential for a dynamically authenticated backend. When set, the call is issued on @@ -726,6 +781,7 @@ def model_turn( "model_turn: mint requires wire_id_map — minted ids are " "unrestorable on the wire without the recovery map" ) + _raise_if_aborted(cancel_ref, lane) # ONE config fetch per plant call feeds both live per-call flags — a # registry hot-reload cannot hand the replay gate and the attach gate # different config generations within a single request. @@ -759,6 +815,10 @@ def model_turn( ) attempt = 0 while True: + # Last read before the wire — it covers everything the entry read is + # too early to see: the lowering, and on a delegated-auth alias the + # credential resolve, which can block. + _raise_if_aborted(cancel_ref, lane) # ``create_streaming`` stays OUTSIDE the try: every adapter issues # the HTTP request eagerly in its body (inside the SDK's own # request-level retry), so an exception from it is a request-time @@ -805,9 +865,10 @@ def model_turn( if delay > 0: time.sleep(delay) if bool(getattr(cancel_ref, "aborted", False)): - # The deadline abandoned this worker while it was backing - # off — die with the original failure instead of - # resurrecting the request from an abandoned thread. + # The deadline abandoned this worker while it was backing off. + # The loop-top read would stop the re-issue anyway; this arm + # exists to die with the ORIGINAL transport failure rather than + # the abandonment error, so the cause of the death survives. raise raw_calls: list[dict[str, Any]] = list(result.tool_calls or []) diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 8494fb81..51aab631 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -377,15 +377,28 @@ class _CancelRef(list[Any]): @property def aborted(self) -> bool: - """Whether this ref's stream must not be resurrected. + """Whether this ref's call must not proceed or be resurrected. - ``model_turn`` consults ``cancel_ref.aborted`` before re-issuing a - request after a mid-drain transport failure (the deadline daemon's - :class:`~turnstone.core.deadline.StreamAbortRef` contract): a - stream that died because :meth:`ChatSession.cancel` closed it — or - because it belongs to a superseded generation — must not be - resurrected behind the user's Stop; the failure surfaces and the - caller's own cancel check turns it into ``GenerationCancelled``. + ``model_turn`` consults ``cancel_ref.aborted`` before every + dispatch and again before re-issuing after a mid-drain transport + failure (the deadline daemon's + :class:`~turnstone.core.deadline.StreamAbortRef` contract): a call + the user's Stop abandoned — or one belonging to a superseded + generation — issues no request, and a stream that died because + :meth:`ChatSession.cancel` closed it is not resurrected behind + that Stop; the failure surfaces and the caller's own cancel check + turns it into ``GenerationCancelled``. + + Deliberately the same two conditions as :meth:`_check_cancelled` + — provided both are asked about the same generation, which on the + compaction lane they are (this ref and that call carry the same + ``my_generation``). Compaction depends on the pairing: + ``_summarize_once``'s handler calls ``_check_cancelled`` before it + reads the error, which is what converts ``model_turn``'s + pre-dispatch raise into a cancelled compaction instead of a red + failure row. Widening this predicate without widening that one, + or pairing a ref with a check on a different generation, breaks + the translation. """ return self._session._cancel_event.is_set() or self._superseded()