From 90e55f92ca218879605353d1f5a04b6a7670c98d Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Wed, 5 Aug 2026 20:15:44 -0700 Subject: [PATCH] docs(832): shorten the branch's comments to their constraints Comment-only sweep over the diff's prose: origin archaeology, next-line narration, and review-thread talk go; each surviving comment states the constraint the code cannot show, re-wrapped to the file's width. The ruled-behavior restatements in the parity transforms and the contract docstrings (eager append, cancel-predicate pairing, carry ownership, the plant call's carve-outs) keep every named invariant. --- tests/_parity_832.py | 60 ++- tests/_session_helpers.py | 34 +- tests/test_832_parity.py | 48 ++- tests/test_cancel.py | 40 +- tests/test_midstream_retry.py | 63 ++-- tests/test_model_provider_obo.py | 22 +- tests/test_think_tag_split.py | 59 ++- turnstone/core/model_turn.py | 150 ++++---- turnstone/core/providers/_protocol.py | 46 +-- turnstone/core/session.py | 520 +++++++++++--------------- turnstone/core/streaming_text.py | 30 +- 11 files changed, 469 insertions(+), 603 deletions(-) diff --git a/tests/_parity_832.py b/tests/_parity_832.py index 330718c0..2867cfdd 100644 --- a/tests/_parity_832.py +++ b/tests/_parity_832.py @@ -1,32 +1,28 @@ """#832 replay-parity harness: scenario table + runner. -The fold's acceptance is a controller-determinism audit: with the plant's -chunk sequence held fixed, the streaming phase must produce an identical -UI event sequence and an identical committed message — modulo the small -set of RULED behavior changes restated (in full) on the transforms in -``test_832_parity.py``. This module is the shared half: the scenario -scripts (one row per chunk-field→UI translation the consumer performs) -and the runner that drives one scenario through the session's streaming -seam, recording everything the turn observably produced. +The audit is controller determinism: with the plant's chunk sequence held +fixed, the streaming phase must produce an identical UI event sequence +and an identical committed message — modulo the RULED behavior changes +restated in full on the transforms in ``test_832_parity.py``. This +module is the shared half: the scenario scripts (one row per +chunk-field→UI translation the consumer performs) and the runner that +drives one through the streaming seam, recording everything the turn +observably produced. Baselines are captured from the PRE-FOLD path (``UPDATE_832_PARITY=1``, -run at a tree where ``session.py`` is byte-identical to pre-fold main — -the fixture commit's history proves it) into ``tests/data/parity_832/``. -The runner adapts to EITHER world by signature (pre-fold -``_stream_response(msgs, my_generation)`` took the prepared wire list; -post-fold ``_stream_response(my_generation)`` prepares inside), so a -recapture at an old tree records real old-world behavior — and capture -mode refuses to write a record whose failure is the harness's own call -shape. The assert mode replays the same scripts through the current -tree and compares against the baseline, applying the ruled transforms. -A mismatch outside a ruled transform is a fold regression. +run at a tree where ``session.py`` is byte-identical to pre-fold main) +into ``tests/data/parity_832/``. The runner adapts to EITHER world by +signature, so a recapture at an old tree records real old-world +behavior, and capture mode refuses to write a record whose failure is +the harness's own call shape. Assert mode replays the same scripts +through the current tree and compares against the baseline, applying the +ruled transforms; a mismatch outside a ruled transform is a regression. The provider fake arms ``cancel_ref`` EAGERLY (a closeable sentinel appended inside ``create_streaming``, before the iterator is returned), -mirroring every real adapter — the post-fold wrapper classifies -creation-vs-midstream failures by that arming, so a fake that skipped -it would exercise only the creation arm and the parity audit would -never reach the mid-stream ladder. +mirroring every real adapter: the wrapper classifies +creation-vs-midstream failures by that arming, so a fake that skipped it +would exercise only the creation arm. """ from __future__ import annotations @@ -155,20 +151,18 @@ def _mask_synth_ids(record: dict[str, Any]) -> dict[str, Any]: def run_scenario(name: str) -> dict[str, Any]: """Drive one scenario through the streaming seam; return the record. - The record is everything the streaming phase observably produced: - the ordered UI events, the committed-message projection, the - mid-stream usage slot, and the exception class if the seam raised. - Deliberately seam-level (the ``_stream_response`` boundary pre-fold, - its wrapper successor post-fold) — full ``send()`` scenarios ride the - ported ladder suites instead. + The record is everything the streaming phase observably produced: the + ordered UI events, the committed-message projection, the mid-stream + usage slot, and the exception class if the seam raised. Deliberately + seam-level at ``_stream_response`` — full ``send()`` scenarios ride + the ported ladder suites instead. Signature-adaptive so ``UPDATE_832_PARITY=1`` at a PRE-fold tree records real old-world behavior: the pre-fold seam was - ``_stream_response(msgs, my_generation) -> dict`` (wire list built - by the caller), the post-fold seam is - ``_stream_response(my_generation) -> ModelTurnResult`` (wire - prepared inside). A harness-shape failure must never be recorded - as behavior — ``write_fixture`` refuses one. + ``_stream_response(msgs, my_generation) -> dict``, the post-fold one + is ``_stream_response(my_generation) -> ModelTurnResult`` (wire + prepared inside). A harness-shape failure must never be recorded as + behavior — ``write_fixture`` refuses one. """ ui = RecordingUI() session = make_session(ui=ui) diff --git a/tests/_session_helpers.py b/tests/_session_helpers.py index 73c03ef7..16446649 100644 --- a/tests/_session_helpers.py +++ b/tests/_session_helpers.py @@ -1,14 +1,13 @@ """Shared session-test helpers. The minimal ``ChatSession`` factory, the ``SessionUIBase`` no-op/recording -subclasses, and — since #832 — the tree's standard streaming provider -fakes (``make_result`` / ``arm_session`` / ``scripted_provider`` / -``ArmedHandle``, at the bottom): every suite that drives the streaming -seam imports them from here so the eager-arming contract lives in one -place. Hoisting keeps callers from drifting on the defaults — the one -deliberate exception, ``test_model_registry.py``'s ``_make_session``, -takes a different signature (registry / model_alias / reasoning_effort -+ ``_FakeUI``) and is NOT a candidate for sharing this helper. +subclasses, and the tree's standard streaming provider fakes +(``make_result`` / ``arm_session`` / ``scripted_provider`` / +``ArmedHandle``, at the bottom): every suite driving the streaming seam +imports them from here, so the eager-arming contract lives in one place. +The one deliberate exception, ``test_model_registry.py``'s +``_make_session``, takes a different signature (registry / model_alias / +reasoning_effort + ``_FakeUI``) and is NOT a candidate for sharing. Module is named with a leading underscore so pytest doesn't try to collect it as a test file — it's an importable utility, not a test. @@ -485,11 +484,10 @@ def make_result( producer: str = "openai-compatible", wire_msgs: list[dict[str, Any]] | None = None, ) -> ModelTurnResult: - """A ``ModelTurnResult`` shaped like the streaming wrapper's return — + """A ``ModelTurnResult`` shaped like the streaming wrapper's return, for tests that only need "a turn happened" and patch - ``_stream_response`` wholesale. The Turn and the ``tool_calls`` - mirror are built from the same dicts, preserving the #825 pairing - invariant fakes must not break.""" + ``_stream_response`` wholesale. Turn and ``tool_calls`` mirror are + built from the same dicts, preserving the #825 pairing invariant.""" calls = list(tool_calls or []) tc_tuple = tuple( ToolCall( @@ -533,12 +531,12 @@ def arm_session( Each ``create_streaming`` call serves the next element of *streams*: an iterable/generator is armed (a closeable sentinel appended to ``cancel_ref`` — the eager append every real adapter performs, which - the fold's creation-vs-midstream classifier keys on) and returned to - be consumed once; an EXCEPTION instance is raised at create time - WITHOUT arming — a creation-phase failure the per-lane ladder owns. - Calls beyond the script fail loudly (the pre-fold lax consumer used - to absorb an exhausted iterator as a silent empty turn; the strict - finish gate rejects that now, so an under-scripted test must say so). + the creation-vs-midstream classifier keys on) and returned to be + consumed once; an EXCEPTION instance is raised at create time WITHOUT + arming, a creation-phase failure the per-lane ladder owns. Calls + beyond the script fail loudly: the strict finish gate rejects an + exhausted iterator rather than absorbing it as a silent empty turn, + so an under-scripted test must say so. Title generation is latched off — with a provider-LEVEL fake the best-effort title lane would otherwise consume the first script diff --git a/tests/test_832_parity.py b/tests/test_832_parity.py index 9ec72548..c9690ea7 100644 --- a/tests/test_832_parity.py +++ b/tests/test_832_parity.py @@ -1,19 +1,18 @@ """#832 replay-parity pins: fixed chunk scripts ⇒ identical observable turn. Baseline fixtures under ``tests/data/parity_832/`` were captured from the -pre-fold streaming path (``UPDATE_832_PARITY=1``; the capturing commit's -``session.py`` is byte-identical to pre-fold main, which is what makes -them THE old-world record). Each test replays a scenario on the current +pre-fold streaming path (``UPDATE_832_PARITY=1`` at a tree whose +``session.py`` is byte-identical to pre-fold main), which is what makes +them THE old-world record. Each test replays a scenario on the current tree and compares the full record — UI event sequence, committed-message -projection, mid-stream usage — against the baseline, after applying the +projection, mid-stream usage — against the baseline after applying the transforms below. Each transform IS a ruled #832 behavior change, restated in full where it is applied, so the pin is auditable from this -file alone. A difference outside a ruled transform is a fold regression. +file alone; a difference outside one is a regression. Do not regenerate baselines casually: they encode the old world. A -legitimate regeneration (a ruled delta superseding capture) updates the -matching transform below in the same commit, or the pin loses its -meaning. +legitimate regeneration updates the matching transform below in the same +commit, or the pin loses its meaning. """ from __future__ import annotations @@ -49,11 +48,11 @@ def _apply_ruled_deltas(name: str, baseline: dict[str, Any]) -> dict[str, Any]: if name == "info_postfinish_footer": # RULED (#832): the trailing citations footer enters the COMMITTED # content (conditional fold: non-blank answer, "\n\n" separator — - # the shared drain-side spelling) and streams as content — - # post-carry-flush, so displayed ordering matches committed — + # the shared drain-side spelling) and streams as content + # post-carry-flush, so displayed ordering matches committed, # instead of an ephemeral info bubble that never survived reload. # Consequence accepted with the ruling: web-controlled footer text - # now reaches storage/search/context/export (release-noted). + # reaches storage/search/context/export (release-noted). footer = "Sources:\n- example.com/page" expected["result"]["content"] += "\n\n" + footer events = [e for e in expected["ui_events"] if e != ["info", footer]] @@ -63,9 +62,8 @@ def _apply_ruled_deltas(name: str, baseline: dict[str, Any]) -> dict[str, Any]: elif name == "no_finish_clean_exhaust": # RULED (#832): a stream that exhausts with no finish reason no - # longer commits its partial silently (the pre-fold lax consumer's - # behavior) — it is a mid-stream death: the re-issue ladder - # finalizes the display and re-drives the turn + # longer commits its partial silently — it is a mid-stream death. + # The re-issue ladder finalizes the display and re-drives the turn # (_MID_STREAM_RETRIES times), then the terminal arm # finalizes+discards and the retryable error surfaces. expected["raised"] = "IncompleteStreamError" @@ -116,15 +114,13 @@ class TestDisplayCommitMirror: """The mirror LAW (no old-world baselines): with one chunk script, the DISPLAYED content stream and the COMMITTED content must agree. - Post-fold the drain assembles the committed turn while the consumer - drives the display; these scenarios interleave provider-parsed - ``reasoning_delta`` with buffered content — the combination the - replay-parity grid never scripted, where a live review caught the - two lanes disagreeing (display dropped or relabeled the buffered - tail the commit kept; with a footer the display showed NOTHING - while the commit carried answer + sources). The consumer's - ``close_run`` at the reasoning boundary and ``partial_tag_tail``'s - proper-prefix contract are what hold these together. + The drain assembles the committed turn while the consumer drives the + display; these scenarios interleave provider-parsed + ``reasoning_delta`` with buffered content, where the two lanes can + disagree (display dropping or relabeling the buffered tail the commit + kept, or showing nothing while the commit carries answer + sources). + The consumer's ``close_run`` at the reasoning boundary and + ``partial_tag_tail``'s proper-prefix contract hold them together. """ _USAGE = UsageInfo(prompt_tokens=1, completion_tokens=1, total_tokens=2) @@ -194,10 +190,8 @@ class TestDisplayCommitMirror: StreamChunk(finish_reason="stop"), ], ), - # Round-2 classes: the boundary close must run while an - # INLINE think block is open (the CoT-leak half), and the - # cross-boundary carry must survive state flips (the - # relabel half). + # The boundary close must run while an INLINE think block is + # open, and the cross-boundary carry must survive state flips. ( "open_inline_think_at_reasoning_boundary", [ diff --git a/tests/test_cancel.py b/tests/test_cancel.py index 82e68250..f5062cad 100644 --- a/tests/test_cancel.py +++ b/tests/test_cancel.py @@ -405,10 +405,9 @@ class TestStreamFlushBeforeToolCalls: ) # Two scripted turns: the tool round-trip makes send() loop, and - # the follow-up turn must carry a real finish reason — the old - # exhausted-iterator path was silently committed as an empty turn - # by the pre-fold lax consumer; the strict finish gate (ruled, - # #832) rejects it now. + # the follow-up turn must carry a real finish reason — the strict + # finish gate (RULED, #832) rejects an exhausted iterator that + # used to commit as an empty turn. arm_session( session, stream_content_then_tool(), @@ -462,8 +461,7 @@ class TestStreamAbort: """The per-attempt ref's eager append registers the SDK handle in ``_cancel_stream`` before the first chunk is consumed — the handle ``cancel()`` closes to unblock a stuck read — and send()'s finally - clears it (#832: the shared ref list is gone; the handle slot is - the surviving cancel surface).""" + clears it.""" ui = NullUI() session = _make_session(ui=ui) @@ -555,13 +553,12 @@ class TestStreamAbort: def test_pre_set_cancel_no_mint_no_dispatch(self, tmp_db): """A Stop already set when the streaming phase is entered issues NO request and — on a dynamically authenticated alias — mints NO - credential (#832; the #972 pre-dispatch rule extended to the - interactive path). Pre-fold, ``_try_stream`` resolved the backend - token BEFORE its first cancel check, so an abandoned turn still - paid a mint (on a cache miss: a network round trip under the - cluster-wide advisory lock). Post-fold the resolver runs inside - ``model_turn`` after its entry abort read, and the ladder's own - loop-top check precedes even the lane build.""" + credential (the #972 pre-dispatch rule on the interactive path). + A mint on a cache miss is a network round trip under the + cluster-wide advisory lock, so an abandoned turn must not pay one: + the resolver runs inside ``model_turn`` after its entry abort + read, and the ladder's loop-top check precedes even the lane + build.""" session = _make_session() session._cancel_event.set() resolver = MagicMock(return_value=None) @@ -624,8 +621,7 @@ class TestCancelRef: """The long-lived shared ref is GONE (#832): every model-call site builds a fresh per-attempt, generation-scoped _CancelRef, so a force-cancelled generation's ref reads aborted via supersession. - This pin holds the line against the gen-0 shared instance coming - back through a fixture or a convenience refactor.""" + The pin holds the line against a gen-0 shared instance returning.""" session = _make_session() assert not hasattr(session, "_cancel_ref") @@ -689,10 +685,10 @@ class TestOnFirstAppendHook: class TestForceCancelOrphanNoReissue: """A force-cancelled generation's mid-stream death is never re-issued - on the orphan's behalf (#832 regression pin: the old shared gen-0 ref - read ``aborted`` False after a force-cancel — the successor installs a - fresh unset event and gen 0 never reads superseded — so the retry - machinery would have spent tokens for a generation nobody owns).""" + on the orphan's behalf. A gen-0 ref would read ``aborted`` False + after a force-cancel (the successor installs a fresh unset event and + gen 0 never reads superseded), and the retry machinery would spend + tokens for a generation nobody owns.""" def test_orphan_death_not_reissued_no_ui_finalize(self, tmp_db): ui = NullUI() @@ -711,8 +707,7 @@ class TestForceCancelOrphanNoReissue: def dying_orphan_stream(): yield FakeChunk(content_delta="old ") # Force-cancel: a successor claims the generation (bumped - # counter + fresh UNSET event) while this stream is mid-body — - # exactly the state where the old shared ref read aborted=False. + # counter + fresh UNSET event) while this stream is mid-body. session._claim_generation() raise ConnectionError("connection reset") @@ -1269,8 +1264,7 @@ class TestEffectStatusPersistence: class TestNeverArmedStopLeavesNoRow: def test_stop_during_creation_persists_nothing(self, tmp_db): """A Stop landing while creation is still connecting — nothing - armed, zero tokens streamed — must not write an assistant row: - pre-fold no row existed for a turn that never streamed, and a + armed, zero tokens streamed — must not write an assistant row: a marker-only row would replay to the model as context on every later turn. (An ARMED zero-token Stop still records its marker via record_cancelled_partial — TestCancelDuringStreaming pins diff --git a/tests/test_midstream_retry.py b/tests/test_midstream_retry.py index 30205855..81c6cdc7 100644 --- a/tests/test_midstream_retry.py +++ b/tests/test_midstream_retry.py @@ -2,16 +2,16 @@ A wire death DURING body iteration surfaces after the request already returned its stream handle, so neither the SDK's ``max_retries`` nor the -creation-time per-lane ladder (``_model_turn_with_retry``) ever sees it. These tests drive -``ChatSession.send()`` with scripted provider streams and pin the retry -loop's contract: bounded re-issue on the normalized retryable shape, the -dead attempt finalized client-side before the retry notice -(``stream_end``) and discarded from the server buffers once the backoff -survives the Stop window (``stream_discarded`` — never -``turn_committed``, whose semantics keep the turn buffer), cancel-in- -retry-window abort, exhaustion surfacing the stream-death wording, and the -``_assistant_pending_tokens`` reset that keeps a post-finish blip from -recycling the prior turn's token count. +creation-time per-lane ladder (``_model_turn_with_retry``) ever sees it. +These tests drive ``ChatSession.send()`` with scripted provider streams +and pin the retry loop's contract: bounded re-issue on the normalized +retryable shape, the dead attempt finalized client-side before the retry +notice (``stream_end``) and discarded from the server buffers once the +backoff survives the Stop window (``stream_discarded`` — never +``turn_committed``, whose semantics keep the turn buffer), +cancel-in-retry-window abort, exhaustion surfacing the stream-death +wording, and the ``_assistant_pending_tokens`` reset that keeps a +post-finish blip from recycling the prior turn's token count. Fixture note: tests zero ``_RETRY_BASE_DELAY`` per instance (else each retry pays real exponential backoff). @@ -358,12 +358,10 @@ class TestMidStreamRetry: ui = RecordingUI() session = _make_session(ui) - # Post-fold the retry gate reads the SERVING lane's provider off - # the frame's consumer (the creation-time handoff register is - # gone). FlakyError is retryable only per THIS provider's set — - # the re-issue proves the gate consulted the lane that armed the - # stream, not some global default. (The distinct-fallback-lane - # variant of this contract is exercised through the real fallback + # The retry gate reads the SERVING lane's provider. FlakyError is + # retryable only per THIS provider's set, so the re-issue proves + # the gate consulted the lane that armed the stream, not a global + # default. (The distinct-fallback-lane variant rides the real # walk in test_model_registry's TestSessionFallback.) provider = arm_session( session, @@ -716,9 +714,8 @@ class TestRecreateWindowClassification: session.send("test") # The dead attempt streamed only its safe-flush prefix ("Ha"); - # the splitter carry ("lf an answer") must NEVER surface as a - # late content token — the stale-armed bug flushed it into the - # just-truncated segment behind a duplicate stream_end. + # the splitter carry ("lf an answer") must NEVER surface as a late + # content token behind a duplicate stream_end. assert ui.of("content") == ["Ha"] assert ui.kinds().count("stream_end") == 1 # The full partial (flushed + carry) still reaches history via @@ -740,9 +737,9 @@ class TestRecreateWindowClassification: _good_stream("never reached"), ) # The re-issue walk's PREAMBLE (before any begin_attempt) raises: - # with the dead attempt's ref properly retired this classifies as - # a creation-phase failure and the ORIGINAL stream death is the - # error the operator sees — not the preamble's. + # with the dead attempt's ref retired this is a creation-phase + # failure, so the ORIGINAL stream death is the error the operator + # sees — not the preamble's. real_tracker = session._get_health_tracker calls: list[int] = [] @@ -781,12 +778,11 @@ class TestRecreateWindowClassification: assert not any("Backend stream died mid-response" in e for e in errors) def test_never_arming_adapter_death_classifies_midstream(self, tmp_db): - """An adapter that ignores ``cancel_ref`` (the pre-#832 letter of - the contract) forfeits the health/usage hook duties, but its - mid-stream death must STILL classify as mid-stream: chunks - reached the display, so a creation-classified death would - silently re-issue the same lane and double-render them. The - consumer's saw-chunk fallback is that tripwire.""" + """An adapter that ignores ``cancel_ref`` forfeits the + health/usage hook duties, but its mid-stream death must STILL + classify as mid-stream: chunks reached the display, so a + creation-classified death would silently re-issue the same lane + and double-render them.""" ui = RecordingUI() session = _make_session(ui) provider = arm_session(session) # install the provider shell only @@ -817,8 +813,7 @@ class TestRecreateWindowClassification: """A deterministic lowering failure is a session-data fault: no dispatch, no health record, no fallback walk — the typed ``WirePreparationError`` rides to the fatal formatter's dedicated - branch. (Pre-fold parity: wire prep ran in send() outside the - streaming try and could reach none of them.)""" + branch.""" ui = RecordingUI() session = _make_session(ui) provider = arm_session(session, _good_stream("unreached")) @@ -848,9 +843,9 @@ class TestDebugDumpLatch: """The debug request dump prints once per ``_stream_response`` invocation. RULED (#832): send()'s overflow-recovery re-invocation prints the RE-PREPARED wire — the dump that diagnoses the recovery — - where the pre-fold accident (wire prep hoisted outside the streaming - try) printed only the first. Within one invocation, re-issues and - fallback lanes re-run the passes but never re-dump.""" + where the pre-fold behavior printed only the first. Within one + invocation, re-issues and fallback lanes re-run the passes but never + re-dump.""" def test_reissue_never_redumps(self, tmp_db): session = _make_session(RecordingUI()) @@ -881,7 +876,7 @@ class TestFallbackFailureRedaction: """The fallback-failure info line lands in the browser transcript and persisted event stream — it carries the exception CLASS, never its text (a ConnectError's str can embed a credential-bearing - base_url; same rule as the re-issue log arm).""" + base_url).""" ui = RecordingUI() session = _make_session(ui) registry = MagicMock() diff --git a/tests/test_model_provider_obo.py b/tests/test_model_provider_obo.py index a7adcc98..e1a2a4dc 100644 --- a/tests/test_model_provider_obo.py +++ b/tests/test_model_provider_obo.py @@ -1519,12 +1519,10 @@ class TestModelOboToken: session._mcp_mint_client.mint_app_token_sync.assert_not_called() def test_main_lane_carries_backend_auth_resolver(self) -> None: - """The main loop's lane wires the session's mint resolver — the + """The main loop's lane wires the session's mint resolver; the credential then resolves and binds INSIDE model_turn per attempt, - after its entry abort read (the #972/#832 ordering: a pre-set Stop - mints nothing, pinned in test_cancel; the with_options SDK binding - itself is model_turn's own pinned behavior). This replaces the - retired pin on _try_stream's hoisted once-per-ladder resolve.""" + after its entry abort read (the #972 ordering — a pre-set Stop + mints nothing, pinned in test_cancel).""" sess = MagicMock() sess._registry = None sess._config_store = None @@ -1543,16 +1541,15 @@ class TestModelOboToken: assert lane.backend_auth_resolver is sess._model_backend_auth_token assert lane.alias == "tf" # The session's own sampling knobs override the lane's operator - # rungs (wire parity with the pre-fold loop). + # rungs. assert lane.temperature == 0.5 assert lane.reasoning_effort is None def test_main_lane_never_consults_config_store(self) -> None: """The alias/global sampling rungs must not reach the main loop: - the session's own knobs replace both values ``config_store`` - would feed, so ``_build_main_lane`` omits the store entirely — - a store with configured rungs contributes nothing to the lane - (wire parity with the pre-fold loop, which never consulted it).""" + the session's own knobs replace both values ``config_store`` would + feed, so ``_build_main_lane`` omits the store entirely and a store + with configured rungs contributes nothing to the lane.""" sess = MagicMock() sess._registry = None store = MagicMock() @@ -1576,9 +1573,8 @@ class TestModelOboToken: def test_primary_lane_built_with_session_alias_for_obo(self) -> None: # Regression: the primary lane must carry the session alias, or the # backend-auth resolver can't resolve the OBO token and an - # entra_obo main turn goes out on the static client key. (The - # pre-fold bug lived in _try_stream's missing model_alias kwarg; - # the lane build is the one place the alias enters now.) + # entra_obo main turn goes out on the static client key. The lane + # build is the one place the alias enters. sess = MagicMock() sess._model_alias = "oboagent" ChatSession._model_turn_with_fallback(sess, MagicMock(), lambda wire: wire) diff --git a/tests/test_think_tag_split.py b/tests/test_think_tag_split.py index 719d866f..8b2564fc 100644 --- a/tests/test_think_tag_split.py +++ b/tests/test_think_tag_split.py @@ -1,15 +1,14 @@ """Behavior pins for the interactive think-tag splitting layer. -``turnstone.core.session._StreamTurnConsumer`` (the main loop's -chunk→UI translation, ``model_turn``'s ``on_chunk`` body post-#832) -splits streamed content into content vs reasoning around -````/```` tags, buffering potential partial tags -across chunk boundaries. These tables pin the CURRENT emission -behavior — exact UI token sequence and final displayed content — so -the logic can move into a standalone ``ThinkTagSplitter`` class with -byte-identical output. Every case drives the real chunk consumer end to -end; none reaches into the implementation, so the same rows must stay -green across the extraction. +``turnstone.core.session._StreamTurnConsumer`` (the main loop's chunk→UI +translation, ``model_turn``'s ``on_chunk`` body) splits streamed content +into content vs reasoning around ````/```` tags, +buffering potential partial tags across chunk boundaries. These tables +pin the CURRENT emission behavior — exact UI token sequence and final +displayed content — so the logic can move into a standalone +``ThinkTagSplitter`` class with byte-identical output. Every case drives +the real chunk consumer end to end; none reaches into the +implementation, so the same rows must stay green across the extraction. Pinned rules: @@ -70,9 +69,9 @@ class _TokenRecorderUI: def _drive(chunks, *, show_reasoning=True, capabilities=None): """Drive *chunks* through a bare ``_StreamTurnConsumer`` — the - display-grid seam post-#832 (``_stream_attempt`` is gone; tool_calls - assembly is the drain's job now, so this helper is for display-only - pins — content emission order plus the accumulated displayed text). + display-grid seam. Tool-call assembly is the drain's job, so this + helper serves display-only pins: content emission order plus the + accumulated displayed text. """ session = make_session() session.show_reasoning = show_reasoning @@ -165,9 +164,9 @@ CASES = [ [("reasoning", "y" * 8), ("reasoning", "y" * 12), ("content", "ok")], "ok", ), - # Reasoning-boundary run close (live-caught #832 divergence): a - # buffered content tail must emit as CONTENT when a reasoning_delta - # arrives, exactly as the drain closes its per-run split there. + # Reasoning-boundary run close: a buffered content tail must emit as + # CONTENT when a reasoning_delta arrives, exactly as the drain closes + # its per-run split there. ( "reasoning_boundary_closes_short_content_run", [_c("Short"), StreamChunk(reasoning_delta="(r)"), _FINISH], @@ -273,10 +272,9 @@ def test_scan_tags_off_returns_every_utterance_byte_identical(case): def test_session_consumer_scan_follows_server_parses_reasoning(): """The interactive consumer wires ``scan_tags`` from the SAME capability the drain seam reads (``server_parses_reasoning``), taken off the - ACTIVE lane post-#832 (no more session-level ``_cached_capabilities`` - read at the consumer). With the flag declared, streamed tag text - reaches the UI verbatim as content — it is prose on such a backend, - not a boundary.""" + ACTIVE lane. With the flag declared, streamed tag text reaches the UI + verbatim as content — it is prose on such a backend, not a + boundary.""" from turnstone.core.providers._protocol import ModelCapabilities content, tokens = _drive( @@ -327,12 +325,10 @@ def test_one_shot_equivalent_to_streaming_over_random_chunkings(case): def test_tool_calls_flush_pending_raw_at_current_state(): # Once tool calls begin, buffered text cannot be a partial tag: it - # flushes RAW (no tag scan) at the current in_think state. Tool_calls - # assembly is the drain's job post-#832 (the display consumer only - # flushes the splitter at the boundary), so this one needs the real - # seam — session._stream_response over scripted_provider — to pin the - # display order and the assembled call together, as the fused - # pre-fold consumer did. + # flushes RAW (no tag scan) at the current in_think state. Assembly + # is the drain's job while the consumer only flushes the splitter, so + # this pin drives the real seam to hold the display order and the + # assembled call together. chunks = [ _c("part`` across a run boundary as if it - might still grow, relabeling the next run.""" + ``startswith`` makes the drain carry a finished ```` across + a run boundary as if it might still grow, relabeling the next run.""" @pytest.mark.parametrize( ("text", "tail"), @@ -397,9 +392,9 @@ class TestCloseRun: def test_partial_tag_tail_is_returned_not_held(self): # The carry's lifetime belongs to the RUN OWNER: the splitter's - # own pending would be re-read under a flipped in_think (the - # round-2 relabel defect), so close_run hands the tail back and - # clears its buffer. + # own pending would be re-read under a flipped in_think and + # relabeled, so close_run hands the tail back and clears its + # buffer. sp, events = self._splitter() sp.feed("Ans bool: """THE inline tag-scan gate — the single spelling of the #978 rule. Scan ```` tags out of the content stream unless the lane's - capabilities declare the server already segregates reasoning - (``server_parses_reasoning``). A lane with no declared capabilities — - or no lane yet — keeps the passthrough-server default: scan. Shared - by the drain seam and the interactive display splitter so the two - interpretations of one stream cannot disagree. + capabilities declare that the server segregates reasoning itself + (``server_parses_reasoning``); no capabilities, or no lane yet, keeps + the passthrough-server default: scan. Shared by the drain seam and + the interactive display splitter so the two readings of one stream + cannot disagree. """ return ( lane is None or lane.capabilities is None or not lane.capabilities.server_parses_reasoning @@ -792,18 +789,17 @@ class ModelTurnResult: *finish_reason* / *usage* are transport facts, not trajectory content — which is why they ride the result, not the Turn. - *wire_msgs* is the exact lowered list handed to ``create_streaming`` - (post every pass, including the caller's ``prepare_wire``) — the - main loop's token-table calibration reads it so the chars-per-token - estimate is computed against what the provider actually counted, - surviving lowerings the caller cannot see (#832; the successor of - the session's ``_wire_msgs`` message-dict carrier). + *wire_msgs* is the exact lowered list handed to ``create_streaming``, + after every pass including the caller's ``prepare_wire`` — the main + loop's token-table calibration reads it so chars-per-token is + computed against what the provider actually counted, lowerings the + caller cannot see included. - *producer* is the SERVING lane's provider name — the same identity - stamped on ``turn.native`` when a native lane exists, carried - separately so a native-less turn still records who produced it (the - storage row's ``producer`` column; pre-fold this read the session's - PRIMARY binding and mislabeled fallback-served turns). + *producer* is the SERVING lane's provider name (the storage row's + ``producer`` column) — the identity stamped on ``turn.native`` when a + native lane exists, carried separately so a native-less turn still + records who produced it and a fallback-served turn is not labeled + with the primary binding. """ turn: Turn @@ -840,14 +836,13 @@ def _tee_chunks( ) -> Iterator[StreamChunk]: """Surface each chunk to *on_chunk* before the drain accumulates it. - Sits UPSTREAM of ``drain_stream``'s own ``transport_guarded`` wrapper, so - the callback observes exactly the chunk sequence the assembler consumes — - no more (transport deaths raise at ``next()`` and never reach the - callback) and no fewer (a callback raise at chunk N, the cancellation - path, discards N from display and assembly alike, matching the - interactive loop's check-before-dispatch ordering). The callback's - exceptions propagate untouched: ``GenerationCancelled`` is a - ``BaseException``, invisible to the drain-retry arm by design. + Sits UPSTREAM of ``drain_stream``'s ``transport_guarded`` wrapper, so + the callback sees exactly the sequence the assembler consumes: no + more (a transport death raises at ``next()`` and never reaches the + callback) and no fewer (a callback raise at chunk N — the + cancellation path — discards N from display and assembly alike). + Callback exceptions propagate untouched; ``GenerationCancelled`` is a + ``BaseException`` and so invisible to the drain-retry arm. """ for sc in chunks: on_chunk(sc) @@ -968,34 +963,33 @@ def model_turn( the red-error path. *prepare_wire* is the caller's OWN deterministic lowering, 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 (#832), 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 - exact post-``prepare_wire``, post-attach list that went to the wire is - returned as ``ModelTurnResult.wire_msgs`` for caller-side calibration. + 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 + *deferred_names* passes through to ``create_streaming``: the tool-search deferred-loading set is per-call state (it grows as the - session discovers tools), which is why it is a parameter here and not - a ``ModelLane`` field. + session discovers tools), so it is a parameter and not a + ``ModelLane`` field. *on_chunk* is the streaming surface (#832): each normalized - :class:`StreamChunk` is surfaced to the caller as it arrives — via a - tee UPSTREAM of the drain, so the callback sees exactly the sequence - the assembler consumes — and the fully assembled result is returned as - usual. Chunk→UI translation is entirely the caller's business; the - canonical Turn always comes from the drain's assembly, never from - anything the callback accumulated. A callback raise aborts the call - with that exception (the interactive cancellation path rides this: - ``GenerationCancelled`` is a ``BaseException`` and passes the retry - arm untouched). **With *on_chunk* present the mid-stream drain retry - below is DISABLED** — the third and final policy carve-out: a - partially-surfaced stream must never be silently re-issued behind a - UI that already rendered its tokens; the streaming caller owns - re-issue, finalizing its display per attempt. + :class:`StreamChunk` reaches the caller as it arrives — via a tee + UPSTREAM of the drain, so the callback sees exactly the sequence the + assembler consumes — and the assembled result is returned as usual. + Chunk→UI translation is the caller's business; the canonical Turn + always comes from the drain's assembly, never from anything the + callback accumulated. A callback raise aborts the call with that + exception; ``GenerationCancelled``, being a ``BaseException``, passes + the retry arm untouched. **With *on_chunk* present the mid-stream + drain retry below is DISABLED** — the third policy carve-out: a + partially-surfaced stream must not be silently re-issued behind a UI + that already rendered its tokens, so the streaming caller owns + re-issue. Raises whatever the provider raises — retry/deadline/fallback policy is the caller's — EXCEPT transient mid-stream deaths: a failure the @@ -1007,12 +1001,10 @@ def model_turn( wire blip; this loop is that retry's new home (request-time failures still get the SDK's own policy inside ``create_streaming`` and propagate unchanged). A *prepare_wire* raise is the one re-typed - exception: it surfaces as :class:`WirePreparationError` (original as - ``__cause__``) so ladders can keep a caller-data fault out of backend - health and fallback walks. An - aborted *cancel_ref* suppresses retries — a deadline that closed the - stream must not have the request resurrected behind its back — and, - read before each dispatch, raises + exception, surfacing as :class:`WirePreparationError` with the + original as ``__cause__``. An aborted *cancel_ref* suppresses + retries — a deadline that closed the 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). @@ -1046,10 +1038,8 @@ def model_turn( try: wire = prepare_wire(wire) except Exception as prep_err: - # Deterministic caller-side lowering over the caller's own - # history — never a backend signal. Typed so the retry and - # fallback ladders can refuse to record health or walk aliases - # over it (each lane would re-run the same failing passes). + # A caller-data fault, never a backend signal — typed so the + # retry and fallback ladders cannot treat it as one. raise WirePreparationError(str(prep_err)) from prep_err wire = maybe_attach_vllm_chat_reasoning(wire, lane.provider, lane.registry, lane.alias, cfg=cfg) # The effort assignment scheme's lower rungs: explicit relay → lane @@ -1074,8 +1064,8 @@ def model_turn( if resolved_backend_auth else lane.client ) - # A partially-surfaced stream is never silently re-issued (see the - # on_chunk docstring section) — the streaming caller owns re-issue. + # A partially-surfaced stream is never silently re-issued — the + # streaming caller owns re-issue. drain_retries = 0 if on_chunk is not None else _DRAIN_RETRIES attempt = 0 while True: diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py index 033264db..acbe71b2 100644 --- a/turnstone/core/providers/_protocol.py +++ b/turnstone/core/providers/_protocol.py @@ -99,11 +99,9 @@ def merge_usage(acc: UsageInfo | None, new: UsageInfo) -> UsageInfo: never mutates ``new`` (the provider's object). Serves :func:`drain_stream` (the one assembler) and the interactive - ``on_chunk`` consumer's live re-projection alike — the streaming - callback re-projects the merged accumulator into the session's - ``_last_usage`` dict on every usage chunk (that dict has mid-stream - readers), so the display lane cannot drift from the assembly on the - merge rule. + ``on_chunk`` consumer, which re-projects the merged accumulator into + the session's ``_last_usage`` dict (read mid-stream) on every usage + chunk — so the display lane cannot drift from the assembly. """ if acc is None: return replace(new) @@ -147,11 +145,10 @@ def accumulate_tool_call_delta( fragment), ``arguments_delta`` concatenates. Returns the (possibly fresh) accumulator entry so callers can hang provider extras off it. - Serves :func:`drain_stream` (the one assembler — post-#832 the - interactive loop assembles here too) and ``GoogleProvider``'s - raw-fidelity capture — every - accumulator in the tree, so the chat loop and the drained lanes - cannot assemble different calls from the same wire stream. + Serves :func:`drain_stream` — the one assembler since #832 folded the + interactive loop into it — and ``GoogleProvider``'s raw-fidelity + capture: every accumulator in the tree, so no two lanes can assemble + different calls from the same wire stream. """ tc = acc.setdefault( tcd.index, @@ -253,12 +250,10 @@ def transport_guarded(chunks: Iterator[StreamChunk]) -> Iterator[StreamChunk]: yield sc -# The trailing-citations fold rule, shared spelling: post-finish info -# (web-search source footers) folds into committed content only onto a -# non-blank answer, joined by this separator. The interactive display -# consumer mirrors the drain's fold with these same two pieces — one -# constant and one predicate — so the streamed and committed renderings -# of a footer cannot drift. +# The trailing-citations fold rule in one place: post-finish info +# (web-search source footers) folds onto a non-blank answer only, joined +# by this separator. The interactive display consumer and the drain +# share both pieces, so a footer cannot render two ways. TRAILING_INFO_SEPARATOR = "\n\n" @@ -437,9 +432,7 @@ def drain_stream( # not exist, folded on, would hand every downstream emptiness check a # truthy footer-only "answer". Blankness, not truthiness; checked # only when a footer exists (footers ride web-search turns only, and - # the strip scan shouldn't tax every drained completion). The gate - # and separator are the shared module-level pair above — the display - # consumer's mirror uses the same two. + # the strip scan shouldn't tax every drained completion). if trailing_info_parts and folds_trailing_info(content): for info in trailing_info_parts: content += TRAILING_INFO_SEPARATOR + info @@ -892,15 +885,14 @@ class LLMProvider(Protocol): are respected. If *cancel_ref* is provided the provider appends the underlying - SDK stream object (which has a ``.close()`` method) EAGERLY — + SDK stream object (which has a ``.close()`` method) EAGERLY: inside this call's body, at HTTP-response time, before the - iterator is even returned (not merely before the first chunk; a - lazily-issued generator adapter would violate this). The append - instant is load-bearing: the caller's creation-vs-midstream - classifier and health recording key on it (#832), and the - eager-append tripwires in ``test_sdk_stream_boundary`` pin it per - adapter. The caller can then close it from another thread to - abort a blocked HTTP read immediately. + iterator is returned — not merely before the first chunk (a + lazily-issued generator adapter would violate this). The + caller's creation-vs-midstream classifier and health recording + key on that instant (#832); ``test_sdk_stream_boundary`` pins it + per adapter. The caller can then close the stream from another + thread to abort a blocked HTTP read immediately. This is the ONLY transport — single-shot callers drain it through :func:`drain_stream` instead of a separate non-streaming entry diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 9ceb4f9c..5a93e281 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -330,8 +330,7 @@ class _CancelRef(list[Any]): """List proxy handed to ``create_streaming`` as its ``cancel_ref``. One FRESH instance per attempt, frame-local to the model-call wrapper - that built it — #832 retired the session-level shared slot - (``test_cancel`` pins ``not hasattr(session, "_cancel_ref")``). + that built it: #832 retired the session-level shared slot. Providers call ``cancel_ref.append(stream_handle)`` eagerly — the HTTP call and registration happen before the iterator is returned to the caller. By overriding ``append`` we update ``ChatSession._cancel_stream`` @@ -339,31 +338,27 @@ class _CancelRef(list[Any]): was created (e.g. cancel during retry backoff), the stream is closed on arrival so the blocked iteration is unblocked. - ``my_generation`` scopes a ref to one generation. EVERY model-call - site now passes its own per-attempt, generation-scoped ref (the main - loop and compaction alike — #832 retired the main loop's long-lived - gen-0 shared instance, whose ``aborted`` was force-cancel-blind: a - successor generation installs a fresh unset event, and gen 0 never - reads superseded). A superseded ref's late-arriving stream — an - abandoned call that passed its boundary check just before a - force-cancel and opened one final zombie call — must neither hijack - ``_cancel_stream`` from the successor generation's live stream nor - keep burning tokens, so the append skips the registration and closes - the stream on arrival. The generation check and the register are two - lockless statements; the residual bytecode-width TOCTOU (successor - claims AND registers between them) is accepted — its harm is one - delayed Stop (closes a dead handle; the event arm still cancels at the - next chunk), not corruption — versus the model-call-width window this - closes. + ``my_generation`` scopes a ref to one generation, and EVERY model-call + site passes its own per-attempt one (main loop and compaction alike). + A long-lived gen-0 ref would be force-cancel-blind: a successor + generation installs a fresh unset event, and gen 0 never reads + superseded. A superseded ref's late-arriving stream — an abandoned + call that passed its boundary check just before a force-cancel and + opened one final zombie call — must neither hijack ``_cancel_stream`` + from the successor generation's live stream nor keep burning tokens, + so the append skips the registration and closes the stream on + arrival. The generation check and the register are two lockless + statements; the residual bytecode-width TOCTOU (successor claims AND + registers between them) is accepted — its harm is one delayed Stop + (closes a dead handle; the event arm still cancels at the next chunk), + not corruption — versus the model-call-width window this closes. ``on_first_append`` fires once, on the first non-superseded append — - i.e. at the adapters' eager HTTP-response-time registration, before - the iterator is returned (the eager-append tripwire in - ``test_sdk_stream_boundary`` pins that timing). It is the main-loop - wrapper's observation point for "the request was accepted": the - health tracker's success record, the creation-vs-midstream retry - classifier, and the per-turn usage-slot resets all key on it (#832). - A superseded arrival does not fire it — an orphan must not record + the adapters' eager HTTP-response-time registration, before the + iterator is returned. It is the wrapper's "request was accepted" + instant: the health success record, the creation-vs-midstream retry + classifier, and the per-turn usage-slot resets all key on it. A + superseded arrival does not fire it — an orphan must not record health or reset the successor's usage slots. """ @@ -422,19 +417,17 @@ class _CancelRef(list[Any]): 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 every - model-call lane now guarantees by construction: compaction and the - main loop alike build a fresh ref per attempt carrying the very - ``my_generation`` their surrounding checks use (#832 closed the - main loop's gen-0 exception). The pairing is load-bearing twice: - compaction's ``_summarize_once`` handler calls ``_check_cancelled`` - before it reads the error, converting ``model_turn``'s pre-dispatch - raise into a cancelled compaction instead of a red failure row, and - the main-loop wrapper's except arms do the same before classifying - a death. Widening this predicate without widening that one, or - pairing a ref with a check on a different generation, breaks the - translation. + Deliberately the same two conditions as :meth:`_check_cancelled`, + provided both are asked about the same generation — which every + model-call lane guarantees by construction, building a fresh ref + per attempt that carries the ``my_generation`` its surrounding + checks use. The pairing is load-bearing twice: compaction's + ``_summarize_once`` handler and the main-loop wrapper's except + arms both call ``_check_cancelled`` before reading the error, + which is what turns ``model_turn``'s pre-dispatch raise into a + cancellation rather than 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() @@ -444,38 +437,35 @@ class _StreamTurnConsumer: One instance per streaming TURN, reset per attempt: display state (splitter carry, spinner latch) is attempt-local, while the instance - itself outlives attempts so the re-issue ladder can read the dead - attempt's partial without riding it on the exception (the old - ``_dead_partial`` hitch-hike — the consumer is frame-local to the - wrapper, so an orphaned generation still cannot poison a successor). + outlives attempts so the re-issue ladder can read the dead attempt's + partial without riding it on the exception. It is frame-local to the + wrapper, so an orphaned generation cannot poison a successor. Deliberately display-side ONLY: the canonical turn is assembled by - ``drain_stream`` inside ``model_turn`` — this class accumulates just - enough to serve the partial-preservation rules (flushed content plus - the splitter's non-think carry) and the live UI grid. Reasoning is - emitted, never accumulated; tool-call deltas only flush the splitter; - ``provider_blocks`` are ignored (assembly owns them). + ``drain_stream`` inside ``model_turn``, and this class accumulates + just enough to serve the partial-preservation rules (flushed content + plus the splitter's non-think carry) and the live UI grid. Reasoning + is emitted, never accumulated; tool-call deltas only flush the + splitter; ``provider_blocks`` are ignored (assembly owns them). The tag-scan posture follows the SAME capability the drain seam reads (``server_parses_reasoning``), taken from the ACTIVE lane — primary or - fallback — so the interactive and drained interpretations of one - stream cannot disagree (the #978 gate, re-homed from the creation-time - handoff register onto the lane). + fallback — so the interactive and drained readings of one stream + cannot disagree. The trailing citations footer is emitted as CONTENT, mirroring the drain's conditional fold (only when the accumulated post-split content is non-blank; ``\\n\\n``-joined) — the #832 ruling that citations - survive into the committed turn; pre-finish info stays an ephemeral - info line exactly as before. + survive into the committed turn. Pre-finish info stays an ephemeral + info line. """ def __init__(self, session: ChatSession, my_generation: int) -> None: self._session = session self._my_generation = my_generation - # No lane until the walk's first begin_attempt resolves one — the - # per-attempt state has exactly ONE initializer (``_reset_attempt``), - # so a first attempt and a re-issue cannot drift, and construction - # costs no resolve_lane walk. + # No lane until the walk's first begin_attempt resolves one: + # per-attempt state has exactly ONE initializer, so a first + # attempt and a re-issue cannot drift. self.lane: ModelLane | None = None self.ref: _CancelRef | None = None self.tracker: BackendHealthTracker | None = None @@ -490,15 +480,13 @@ class _StreamTurnConsumer: self._path1_reasoning = False self._finish_seen = False self._saw_chunk = False - # Content-state text held across a reasoning block (the run-close - # carry) and post-finish footer parts held for the end-of-stream - # fold — both owned HERE, outside the splitter, so in_think flips - # cannot relabel them (mirrors the drain's separate carry). + # The run-close carry and the held footer parts are owned HERE, + # outside the splitter, so an ``in_think`` flip cannot relabel + # them. self._boundary_carry = "" self._trailing_info: list[str] = [] # Per-attempt: a re-issued attempt's usage must never max-merge - # onto the dead attempt's (the accumulator was attempt-local in - # the pre-fold consumer too). + # onto the dead attempt's. self._usage_acc: UsageInfo | None = None self._splitter = ThinkTagSplitter( self._flush_text, scan_tags=lane_scans_inline_reasoning(self.lane) @@ -512,10 +500,9 @@ class _StreamTurnConsumer: ) -> None: """Arm display state for one creation attempt on *lane*. - The usage-slot resets do NOT happen here — they ride - :meth:`on_stream_armed` (the request-accepted instant), so a long - creation ladder or fallback walk never blanks the reconnecting - tab's status bar mid-window. + The usage-slot resets ride :meth:`on_stream_armed` (the + request-accepted instant) instead, so a long creation ladder or + fallback walk never blanks the reconnecting tab's status bar. """ self.ref = ref self.tracker = tracker @@ -526,17 +513,14 @@ class _StreamTurnConsumer: """Pronounce the current attempt dead: nothing about it may leak. The re-issue ladder calls this once a mid-stream death's partial - is captured — from here until the next ``begin_attempt`` there is - NO live attempt, so ``attempt_armed`` reads False. Without it, a - Stop (or a walk-preamble failure) landing in the re-create window - would read the DEAD attempt's armed ref: the cancel arm would - re-finalize discarded display state — the stale splitter carry - emitted as a fresh content token into the just-truncated segment, - with a duplicate ``stream_end`` behind it — and a preamble error - would be classified as another armed death, replacing the - operator-actionable stream-death error. The lane survives: the - ladder's terminal predicate is judged by the lane that actually - armed the dead stream. + is captured: from here until the next ``begin_attempt`` there is + NO live attempt, so ``attempt_armed`` reads False. Otherwise a + Stop or a walk-preamble failure landing in the re-create window + reads the DEAD attempt's armed ref — re-finalizing discarded + display state, or classifying a preamble error as another armed + death and replacing the operator-actionable stream-death error. + The lane survives: the ladder's terminal predicate is judged by + the lane that actually armed the dead stream. """ self.ref = None self._reset_attempt() @@ -544,26 +528,22 @@ class _StreamTurnConsumer: @property def attempt_armed(self) -> bool: """Whether this attempt streamed: handle registered, or any chunk - actually surfaced. The chunk fallback is the tripwire for an - adapter that skips the eager ``cancel_ref`` append (an out-of-tree - adapter written to the pre-#832 letter of the contract): its - mid-stream death must still classify as mid-stream — a - creation-classified death would silently re-issue the same lane - and double-render everything already on screen. Such an adapter - still forfeits ``on_stream_armed``'s duties (health success, - usage-slot resets); the in-tree tripwires pin eager arming.""" + surfaced. The chunk fallback covers an adapter that skips the + eager ``cancel_ref`` append — its mid-stream death must still + classify as mid-stream, since a creation-classified death would + silently re-issue the same lane and double-render what is already + on screen. Such an adapter still forfeits ``on_stream_armed``'s + duties (health success, usage-slot resets).""" return (self.ref is not None and self.ref.armed) or self._saw_chunk def on_stream_armed(self) -> None: """`_CancelRef.on_first_append` — the request-accepted instant. - Carries three duties at exactly the old create-return timing: - the health success record for the serving lane, and the two - per-turn usage-slot resets whose placement guards both the + Three duties: the health success record for the serving lane, and + the two per-turn usage-slot resets, placed here to guard both the stale-usage leak (a post-finish blip can lose the trailing usage chunk, and a stale completion count would be recycled as the next - turn's estimate) and - the reconnect status-bar blackout. + turn's estimate) and the reconnect status-bar blackout. """ s = self._session s._last_usage = None @@ -589,17 +569,15 @@ class _StreamTurnConsumer: self._first_token = False def __call__(self, chunk: StreamChunk) -> None: - # Before the cancel check: the chunk DID arrive, so the attempt - # streamed — the classifier fallback must see that even when this - # very chunk's cancel check aborts the turn. + # Set before the cancel check: the chunk arrived, so the attempt + # streamed even when this very chunk's check aborts the turn. self._saw_chunk = True s = self._session s._check_cancelled(self._my_generation) if chunk.finish_reason: self._finish_seen = True - # Usage re-projects on EVERY usage chunk via THE shared max-merge - # rule — one atomic dict rebind, because the SSE replay preamble + # One atomic dict rebind per usage chunk: the SSE replay preamble # reads this slot from the connection thread mid-stream. if chunk.usage: self._usage_acc = merge_usage(self._usage_acc, chunk.usage) @@ -620,17 +598,12 @@ class _StreamTurnConsumer: if chunk.reasoning_delta: self._stop_spinner_once() # Entering the native-reasoning phase closes the current run - # UNCONDITIONALLY, exactly as the drain does: decided text - # emits at its own state (content-state tail as content, - # in-think tail as reasoning — gating this on in_think left - # an open inline think block's carry to be relabeled as - # displayed ANSWER text by the later state flip, the CoT-leak - # half of the round-2 findings), and only a partial tag - # prefix carries across the block. The carry lives on the - # CONSUMER, not in the splitter's pending — pending is read - # under whatever state later flushes hit, and the in_think - # flip below would relabel a content-state carry as - # reasoning (the other half). + # UNCONDITIONALLY, as the drain does: an open inline think + # block's carry must close at its own state or the later + # state flip relabels it, and only a partial tag prefix + # carries across the block. The carry lives on the CONSUMER, + # not in the splitter's pending, which is read under whatever + # state later flushes hit. self._boundary_carry += self._splitter.close_run() self._splitter.in_think = True self._path1_reasoning = True @@ -644,10 +617,9 @@ class _StreamTurnConsumer: self._path1_reasoning = False self._splitter.in_think = False if self._boundary_carry: - # Content resumed after the reasoning block: the carry - # prefixes the new text so a tag the server split across - # the block reassembles (the drain prefixes its carried - # tail onto the next run the same way). + # The carry prefixes the resumed text so a tag the server + # split across the reasoning block reassembles — the + # drain prefixes its carried tail the same way. self._splitter.feed(self._boundary_carry) self._boundary_carry = "" self._splitter.feed(chunk.content_delta) @@ -658,8 +630,8 @@ class _StreamTurnConsumer: self._stop_spinner_once() if self._boundary_carry: # A partial tag never spans a TOOL boundary (the drain's - # rule): the carry is content-state text — flush it as - # content, whatever in_think says now. + # rule), and the carry is content-state text: flush it as + # content whatever ``in_think`` says now. self._flush_text(self._boundary_carry, False) self._boundary_carry = "" self._splitter.flush_pending() @@ -668,13 +640,11 @@ class _StreamTurnConsumer: if chunk.info_delta: self._stop_spinner_once() if self._finish_seen: - # Post-finish info is the trailing citations footer: HELD - # and folded once at stream end after ALL content - # (:meth:`finish_stream`), exactly the drain's post-loop - # fold. Folding at arrival diverges from the commit when - # a lax gateway emits content after finish — the fold - # gate must judge the FULL answer, and the footer must - # follow it. + # The trailing citations footer is HELD and folded once + # at stream end (:meth:`finish_stream`), like the drain's + # post-loop fold: a lax gateway can emit content after + # finish, and the fold gate must judge the FULL answer + # with the footer behind it. self._trailing_info.append(chunk.info_delta) else: s.ui.on_info(f"{GRAY}{chunk.info_delta}{RESET}") @@ -682,11 +652,11 @@ class _StreamTurnConsumer: # -- partial preservation -------------------------------------------------- def partial_content(self) -> str: - """THE partial-content rule: flushed content, plus the boundary - carry (content-state by construction), plus the splitter's carry - tail when it is content-state (an in-think tail is reasoning and - stays out) — one closure serving the cancel arms and the re-issue - ladder's dead-partial promotion alike.""" + """THE partial-content rule: flushed content, the boundary carry + (content-state by construction), and the splitter's tail only + when it is content-state — an in-think tail is reasoning and + stays out. Serves the cancel arms and the re-issue ladder's + dead-partial promotion alike.""" return ( "".join(self._content_parts) + self._boundary_carry @@ -695,9 +665,9 @@ class _StreamTurnConsumer: def _flush_terminal_carries(self) -> None: """Terminal display flush shared by the finish and cancel arms: - the boundary carry emits as CONTENT (its state was fixed when the - run closed — the drain appends its dangling carry to content the - same way), then the splitter's own pending at the current state.""" + the boundary carry emits as CONTENT — its state was fixed when + the run closed — then the splitter's pending at the current + state.""" if self._boundary_carry: self._flush_text(self._boundary_carry, False) self._boundary_carry = "" @@ -707,15 +677,13 @@ class _StreamTurnConsumer: """End-of-stream display flush: the held carries, then the trailing citations footer. - The drain assembled the canonical content already; without this - the DISPLAYED stream is missing its last ≤MAX_TAG_LEN characters - (the carries), and the footer fold must run AFTER them — once, - over the full answer, with the shared gate and separator — or - the displayed fold diverges from the drain's post-loop fold. - Success-path only, after the trailing Stop re-check — the cancel - arms flush via :meth:`record_cancelled_partial` and drop the - footer (a cancelled turn commits no drained content to fold - onto).""" + Without this the DISPLAYED stream is missing its last + ≤MAX_TAG_LEN characters, and the footer fold must run AFTER them + — once, over the full answer — or it diverges from the drain's + post-loop fold. Success-path only, after the trailing Stop + re-check: the cancel arms flush via + :meth:`record_cancelled_partial` and drop the footer, since a + cancelled turn commits no drained content to fold onto.""" self._flush_terminal_carries() if self._trailing_info and folds_trailing_info("".join(self._content_parts)): for info in self._trailing_info: @@ -727,8 +695,8 @@ class _StreamTurnConsumer: send()'s cancel handler. No-op for a SUPERSEDED generation: an orphan must touch neither the UI nor the shared partial slot. ``tool_calls`` and the native lane are DELIBERATELY omitted — - incomplete calls would orphan their results, and the marker-as- - message contract needs plain content (same rules as ever).""" + incomplete calls would orphan their results, and the + marker-as-message contract needs plain content.""" if self._session._generation != self._my_generation: return content = self.partial_content() @@ -2283,8 +2251,8 @@ class ChatSession: # content-addressed against the turn; same lifecycle as the two above. self._tool_previews: dict[str, tuple[dict[str, Any], Attachment]] = {} # Cooperative cancellation: set from outside to stop generation. - # No long-lived cancel REF: every model-call site builds a fresh - # per-attempt, generation-scoped _CancelRef (#832) — this slot is + # No long-lived cancel REF: every model-call site builds its own + # per-attempt, generation-scoped _CancelRef, and this slot is only # the closeable handle those refs register for cancel(). self._cancel_event = threading.Event() self._cancel_stream: Any = None # closeable SDK stream handle @@ -2296,11 +2264,6 @@ class ChatSession: # the model detached on purpose. close() reaps everything. self._background_shells = BackgroundShellRegistry(on_exit=self._on_background_shell_exit) self._cancelled_partial_msg: dict[str, Any] | None = None - # (The creation-time handoff register — _active_stream_provider / - # _active_stream_caps — is gone: the streaming wrapper's frame - # holds the ACTIVE ModelLane itself, so the retry gate's - # retryable set and the consumer's tag-scan posture read the - # serving lane by construction, fallback walk included. #832) self._pending_retry: str | None = None # True when a fatal exception's text has been persisted to # workstream_config["last_error"] for the coord's inspect/wait @@ -5766,11 +5729,9 @@ class ChatSession: name = type(exc).__name__ - # A wire-preparation failure is the session's own data at fault — - # ``model_turn`` typed it so no ladder recorded health or walked - # fallbacks. Checked FIRST: it is the most specific classification, - # and the overflow text-match below must not claim a lowering error - # whose message happens to mention token limits. + # A wire-preparation failure is the session's own data at fault, not + # the backend's. Checked FIRST: the overflow text-match below must + # not claim a lowering error whose message mentions token limits. if isinstance(exc, WirePreparationError): prep_cause = exc.__cause__ detail = f"{type(prep_cause).__name__}: {prep_cause}" if prep_cause else raw_msg @@ -6226,16 +6187,12 @@ class ChatSession: The session's OWN sampling knobs override the lane's operator-resolved rungs (``dataclasses.replace`` on the frozen - lane): the pre-fold loop passed ``self.temperature`` / - ``self.reasoning_effort`` to the wire verbatim and never consulted - the per-alias config rungs, and a resumed workstream with unset - knobs must keep OMITTING rather than newly picking up alias - config. ``config_store`` is deliberately NOT passed: its only - consumers inside ``resolve_lane`` are the two sampling-knob - resolvers this method overrides, and a dead store read per lane - build would misread as the rungs reaching the main loop. Whether - they SHOULD is a live question — but not this fold's (wire parity - first). + lane): a resumed workstream with unset knobs must keep OMITTING + them rather than picking up per-alias config. ``config_store`` + is deliberately NOT passed — its only consumers inside + ``resolve_lane`` are the two sampling-knob resolvers this method + overrides, so a dead store read per lane build would misread as + those rungs reaching the main loop. """ lane = resolve_lane( provider, @@ -6258,17 +6215,16 @@ class ChatSession: prepare_wire: Callable[[list[dict[str, Any]]], list[dict[str, Any]]], my_generation: int = 0, ) -> ModelTurnResult: - """Run one plant call with lane-swap fallback — the successor of the - stream-creation walk, one ``model_turn`` ladder per lane. + """Run one plant call with lane-swap fallback: one ``model_turn`` + ladder per lane. - Health semantics preserved exactly: success records at the - request-accepted instant (the consumer's ``on_stream_armed`` hook), - failure records HERE, once per lane's whole ladder — and an ARMED - death records neither, because a mid-stream death was never a - creation-health signal (it re-raises straight to the re-issue - ladder in ``_stream_response``; a fallback lane whose stream died - after tokens reached the UI must never be swallowed into - try-the-next-alias, or the turn double-renders). + Success records at the request-accepted instant (the consumer's + ``on_stream_armed`` hook), failure records HERE once per lane's + whole ladder, and an ARMED death records neither — a mid-stream + death is not a creation-health signal, and it re-raises to the + re-issue ladder in ``_stream_response`` rather than being + swallowed into try-the-next-alias, which would double-render the + turn. """ tracker = self._get_health_tracker() primary_lane = self._build_main_lane( @@ -6287,11 +6243,10 @@ class ChatSession: # refusal as backend health and never route it to a static fallback. raise except WirePreparationError: - # Session-data fault, typed by ``model_turn``: the caller's own - # lowering raised. No health record and no fallback walk — - # every lane would re-run the same deterministic passes, and - # N recorded failures would paint a cluster-wide outage over - # a malformed turn in ONE session's history. + # No health record and no fallback walk: every lane would + # re-run the same deterministic passes, and N recorded + # failures would paint a cluster-wide outage over one + # session's malformed history. raise except Exception as primary_err: if consumer.attempt_armed: @@ -6366,21 +6321,18 @@ class ChatSession: fb_lane, fb_tracker, consumer, prepare_wire, my_generation ) except (BackendAuthUnavailableError, WirePreparationError): - # Same two verbatim-forward classes as the primary walk: an - # auth refusal is fail-closed policy, a wire-preparation - # failure is the caller's data bug — neither is this - # fallback's health signal. + # An auth refusal is fail-closed policy and a wire-preparation + # failure is the caller's data bug: neither is this fallback's + # health signal. raise except Exception as fb_err: if consumer.attempt_armed: raise if fb_tracker: fb_tracker.record_failure() - # Class name only in the UI line — the same rule as the - # re-issue log arm: a ConnectError's text can carry a - # credential-bearing base_url verbatim, and this string lands - # in the browser transcript and persisted event stream. The - # full detail goes to the server log. + # Class name only in the UI line: a ConnectError's text can + # carry a credential-bearing base_url, and this string lands in + # the browser transcript and the persisted event stream. log.warning( "fallback.failed", alias=alias, @@ -6421,20 +6373,18 @@ class ChatSession: """One lane's creation ladder around ``model_turn``. Per-attempt state is a FRESH generation-scoped ``_CancelRef`` whose - ``on_first_append`` hook marks the request-accepted instant — the - creation-vs-midstream classifier: an ARMED attempt's death re-raises - immediately to the re-issue ladder (its tokens may be on screen; a - silent same-lane retry would double-render), while an unarmed - failure is a creation failure and retries here. The old shared - gen-0 ref is gone — a force-cancelled generation's ref now reads - ``aborted`` via supersession, so ``model_turn`` refuses dispatch for - orphans by construction. + ``on_first_append`` hook marks the request-accepted instant, which + classifies creation vs mid-stream: an ARMED attempt's death + re-raises to the re-issue ladder (its tokens may be on screen, and + a silent same-lane retry would double-render), while an unarmed + failure is a creation failure and retries here. Generation-scoping + makes an orphan's ref read ``aborted`` via supersession, so + ``model_turn`` refuses dispatch for it. Sampling knobs ride the lane (see ``_build_main_lane``); the credential resolves INSIDE ``model_turn`` per attempt, after its - entry abort read — a Stop set before the turn no longer mints a - token on a dynamically authenticated alias (#832; the #972 rule - extended to the interactive path). + entry abort read, so a Stop set before the turn mints no token on + a dynamically authenticated alias (#972). """ raw_url = str(getattr(lane.client, "base_url", getattr(lane.client, "_base_url", "?"))) safe_url = raw_url.split("?")[0] # strip query params (may contain keys) @@ -6464,17 +6414,15 @@ class ChatSession: on_chunk=consumer, ) except Exception as e: - # A Stop — or supersession — is never a backend failure: - # convert BEFORE any classification, the same pre-read - # translation compaction's handler performs. This one line - # turns ``model_turn``'s pre-dispatch DeadlineCancelledError, - # a post-``cancel()`` transport death, and an orphaned - # generation's error into ``GenerationCancelled``. + # A Stop — or supersession — is never a backend failure, so + # convert BEFORE any classification: this turns a + # pre-dispatch DeadlineCancelledError, a post-``cancel()`` + # transport death, and an orphan's error into + # ``GenerationCancelled``. self._check_cancelled(my_generation) if consumer.attempt_armed: - # Mid-stream death — the re-issue ladder owns armed - # deaths (UI finalize → notice → backoff → discard → - # full re-create), on every lane. + # The re-issue ladder owns armed deaths (UI finalize → + # notice → backoff → discard → full re-create). raise ename = type(e).__name__ cause_name = ( @@ -7339,11 +7287,9 @@ class ChatSession: zero_budget_compact_attempts = 0 while True: self._check_cancelled(my_generation) - # Wire preparation (and the debug request dump) live in the - # streaming wrapper's ``prepare_wire`` closure now — run - # inside ``model_turn`` per attempt, so a mid-retry rebind - # re-prepares by construction and this frame never holds a - # wire copy. + # Wire preparation and the debug dump live in the streaming + # wrapper's ``prepare_wire`` closure — this frame holds no + # wire copy, so a mid-retry rebind re-prepares by itself. # Reset the per-turn inflight buffers BEFORE entering # the streaming phase so the SSE refresh-resume snapshot @@ -7409,10 +7355,9 @@ class ChatSession: finally: # Only clear if this generation is still active — # an orphaned thread must not clobber a newer stream. - # (The per-attempt cancel refs die with their frames — - # #832 — but this slot is the handle cancel() closes, - # and a completed turn's dead handle must not linger - # into tool execution.) + # This slot is the handle cancel() closes: a completed + # turn's dead handle must not linger into tool + # execution. if self._generation == my_generation: self._cancel_stream = None self.ui.on_thinking_stop() @@ -7422,19 +7367,14 @@ class ChatSession: return # The wire fold the provider ACTUALLY counted rides the - # result (``wire_msgs`` — a mid-retry rebind re-prepared it - # inside the streaming wrapper, invisibly to this frame), - # keeping the calibration char count aligned with what the - # provider counted. Frame-owned, so a superseding - # generation cannot alias it; a fake result without it - # falls through to the on-the-fly re-fold inside - # ``_update_token_table``. + # result: a mid-retry rebind re-prepared it inside the + # streaming wrapper, invisibly to this frame. A fake + # result without it re-folds inside _update_token_table. self._update_token_table(msgs=result.wire_msgs) self._print_status_line() # Report usage for EVERY API call - # The canonical Turn — minted tool ids, finalized native - # lane, and an ACCURATE producer (the serving lane's, so a - # fallback-served turn no longer wears the primary's name - # and an in-memory fork no longer decodes producer=""). + # The canonical Turn: minted tool ids, finalized native + # lane, and the SERVING lane's producer, so a + # fallback-served turn is not labeled with the primary's. self.messages.append(result.turn) # Clear per-turn inflight buffers — the assistant # message is now in the history list a refresh would @@ -7452,8 +7392,7 @@ class ChatSession: # Log assistant message to conversation history. ONE # binding for the call list: the persisted mirror and the - # executed set below must be the same value by - # construction, not by coincidence. + # executed set below must be the same value. content = result.content tool_calls = result.tool_calls or None native = result.turn.native @@ -8253,20 +8192,17 @@ class ChatSession: The plant call is ONE ``model_turn`` invocation per attempt (creation + drain fused), reached through the lane-swap fallback walk; chunk→UI translation lives in the frame's - :class:`_StreamTurnConsumer`, handed to ``model_turn`` as - ``on_chunk``. Wire preparation is the ``prepare_wire`` closure — - the session's own lowering composed after the seam passes — so a - mid-retry registry rebind needs no explicit re-prepare: the next - attempt re-runs the closure against the refreshed binding by - construction. + :class:`_StreamTurnConsumer`, handed over as ``on_chunk``. Wire + preparation is the ``prepare_wire`` closure, so a mid-retry + registry rebind needs no explicit re-prepare: the next attempt + re-runs the closure against the refreshed binding. A wire death DURING body iteration surfaces after the request was - accepted (the attempt's ``_CancelRef`` armed), so neither the - SDK's ``max_retries`` nor the per-lane creation ladder ever sees - it; ``model_turn``'s own drain retry is DISABLED on this path (a - partially-surfaced stream is never silently re-issued). This loop - finalizes the dead attempt in every UI consumer, then re-issues - the whole turn (the APIs cannot resume a generation) up to + accepted (the attempt's ``_CancelRef`` armed), so neither the SDK's + ``max_retries`` nor the per-lane creation ladder ever sees it, and + ``model_turn``'s own drain retry is DISABLED here. This loop + finalizes the dead attempt in every UI consumer, then re-issues the + whole turn (the APIs cannot resume a generation) up to ``_MID_STREAM_RETRIES`` times. Ladder stacking: a 3-way stack — each re-issue runs the full @@ -8274,15 +8210,14 @@ class ChatSession: fallback-chain passes, so a persistently transient-shaped failure burns (_MID_STREAM_RETRIES + 1) x ((_MAX_RETRIES + 1) + fallback passes) calls before the terminal error surfaces. Both inner - layers are the pre-fold creation path (task_agent's ``_api_call`` - documents the equivalent 2-way stack); every layer stops - immediately on a non-retryable class. + layers are the creation path's own ladders (task_agent's + ``_api_call`` documents the equivalent 2-way stack); every layer + stops immediately on a non-retryable class. """ attempt = 0 # The latest non-empty dead attempt's flushed text. Wrapper-LOCAL # (read off the frame's consumer, never a session slot), so an - # orphaned superseded generation cannot poison a live generation's - # preservation. + # orphaned generation cannot poison a live one's preservation. dead_partial = "" # The armed death whose re-issue is in progress; when the RE-CREATE # phase fails with an unarmed error, the original death is the one @@ -8295,13 +8230,11 @@ class ChatSession: def _prepare(lowered: list[dict[str, Any]]) -> list[dict[str, Any]]: """The main loop's ``prepare_wire``: system prepend + the session lowering passes, plus the debug request dump behind a - once-per-invocation latch — re-issues and fallback lanes + once-per-invocation latch, so re-issues and fallback lanes re-run the passes but not the dump. send()'s overflow - recovery calls ``_stream_response`` again and prints again: + recovery calls ``_stream_response`` again and prints again — post-compaction the wire CHANGED, and the re-prepared dump is - the one that diagnoses the recovery (a named #832 delta — the - pre-fold print-once-per-send was an accident of wire prep - living outside the streaming try).""" + the one that diagnoses the recovery (a named #832 delta).""" nonlocal debug_printed wire = self._prepare_wire_messages([*self.system_messages, *lowered]) if self.debug and not debug_printed: @@ -8331,13 +8264,11 @@ class ChatSession: and not dead_partial ): # Nothing ever streamed this send: a Stop in the - # creation/walk window with no prior armed death. - # Pre-fold, the first creation ran OUTSIDE the promote - # arm and no assistant row was written for a turn that - # never streamed — keep that: a marker-only row would - # replay to the model as context on every later turn - # (round-2 finding; an ARMED zero-token Stop still - # records its marker via record_cancelled_partial). + # creation/walk window with no prior armed death. A turn + # that never streamed writes no assistant row — a + # marker-only row would replay to the model as context on + # every later turn. (An ARMED zero-token Stop still + # records its marker via record_cancelled_partial.) return cur = self._cancelled_partial_msg if cur is None or (not cur.get("content") and dead_partial): @@ -8351,9 +8282,8 @@ class ChatSession: result = self._model_turn_with_fallback(consumer, _prepare, my_generation) # A Stop that raced the trailing-metadata window: cancel() # closed the stream and the drain's post-finish tolerance - # ended it CLEANLY — without this re-check the turn would - # commit as complete and its tool calls would execute - # despite the Stop. + # ended it CLEANLY, so without this re-check the turn + # commits as complete and its tool calls execute. self._check_cancelled(my_generation) consumer.finish_stream() return self._finalize_stream_result(result) @@ -8379,12 +8309,11 @@ class ChatSession: new_dead = consumer.partial_content() if armed else "" dead_partial = new_dead or dead_partial if armed: - # The attempt is dead and its partial captured — from - # here until the next ``begin_attempt`` there is no - # live attempt. Without this, a Stop or a - # walk-preamble failure landing in the re-create - # window reads the DEAD attempt's armed state (see - # ``end_attempt``). + # The partial is captured, so there is no live attempt + # until the next ``begin_attempt``: without this, a + # Stop or a walk-preamble failure landing in the + # re-create window reads the DEAD attempt's armed + # state (see ``end_attempt``). consumer.end_attempt() if self._generation != my_generation: # Superseded (force-cancel started a newer generation): @@ -8394,20 +8323,18 @@ class ChatSession: raise if not armed: # Creation-phase failure: the walk already ran its full - # ladder + fallbacks. Mid re-issue it must not MASK the - # original stream death (a closed-client re-create + # ladder + fallbacks. Mid re-issue it must not MASK + # the original stream death (a closed-client re-create # surfaces as a retryable APIConnectionError and would - # replace the operator-actionable wording) — EXCEPT the - # classes that carry their own remediation: a - # deterministic overflow surfaces as ITSELF so send()'s + # replace the operator-actionable wording) — EXCEPT + # the classes carrying their own remediation: an + # overflow surfaces as ITSELF so send()'s # compact-and-retry arm can recover the turn, and an - # auth refusal / wire-preparation fault surfaces as + # auth refusal or wire-preparation fault surfaces as # ITSELF so the fatal formatter's dedicated branch - # renders (an OBO mint dying mid-turn is a config - # outage, not a network flap — the walk forwards both - # verbatim for exactly that reason). - # Class name only in the log — a ConnectError's text can - # carry a credential-bearing base_url verbatim. + # renders. Class name only in the log — a + # ConnectError's text can carry a credential-bearing + # base_url verbatim. if ( last_stream_death is None or isinstance(e, (BackendAuthUnavailableError, WirePreparationError)) @@ -8502,10 +8429,9 @@ class ChatSession: # clients whose connection config changed — the # in-flight read then dies with a ReadError and # self.client is CLOSED. Generation-gated (two compares - # when nothing changed) and cheap; the re-prepare the - # old path ran on a changed binding is now implicit — - # the next attempt's ``prepare_wire`` closure runs - # against whatever binding the walk resolves. + # when nothing changed); the next attempt's + # ``prepare_wire`` closure re-prepares against whatever + # binding the walk resolves. self._refresh_model_from_registry() except GenerationCancelled: # A Stop landing in the backoff window aborts the turn @@ -8518,13 +8444,12 @@ class ChatSession: """Post-drain policies for a COMPLETED interactive turn. The ``length`` partial-tool-call drop is harness policy, not - assembly: the drain keeps everything it accumulated, and this is - where the interactive lane discards calls whose JSON arguments a - truncation cut mid-string — executing them would dispatch garbage - (the sub-agent loop's policy differs: it stops the run instead). - The rebuilt turn keeps its reasoning synth but drops the orphan - native client tool blocks, via the SAME shared finalize the - assembly used (``has_tool_calls=False`` arm) — no private strip. + assembly: the drain keeps what it accumulated, and the interactive + lane discards calls whose JSON arguments a truncation cut + mid-string, since executing them would dispatch garbage (the + sub-agent loop stops the run instead). The rebuilt turn keeps its + reasoning synth but drops the orphan native client tool blocks via + the SAME shared finalize the assembly used — no private strip. """ finish_reason = result.finish_reason if finish_reason == "length": @@ -8567,13 +8492,11 @@ class ChatSession: self.ui.on_error("Warning: response blocked by content filter.") # Non-destructive integrity signal: the length-guard above drops - # tool calls only on ``finish_reason == "length"``, so a model that - # emits invalid-JSON arguments with a ``stop``/``tool_calls`` - # finish reason commits them verbatim. The canonical Turn stays a - # faithful record (the wire copy is legalized by - # ``lowering.sanitize_tool_call_arguments``); this only flags the - # model-quality problem at the moment it happens, not merely as a - # downstream wire legalization on every replay. + # tool calls only on ``finish_reason == "length"``, so invalid-JSON + # arguments with a ``stop``/``tool_calls`` finish reason commit + # verbatim. The canonical Turn stays a faithful record (the wire + # copy is legalized by ``lowering.sanitize_tool_call_arguments``); + # this flags the model-quality problem where it happens. for tc in result.tool_calls: raw_args = tc["function"].get("arguments") if not wire_valid_arguments(raw_args): @@ -8763,9 +8686,7 @@ class ChatSession: redundant ``_prepare_wire_messages`` walk and ensures the char count matches the bytes the provider counted. When *msgs* is None the caller didn't have one (fake results, direct calls) — - fall back to folding on the fly. (The old leading - ``assistant_msg`` parameter was never read by the body and is - gone — #832.) + fall back to folding on the fly. """ if not self._last_usage: return @@ -9146,11 +9067,10 @@ class ChatSession: # landed), so cancel() aborts the blocked read instead # of waiting out a whole model call — the force-stop # orphan window collapses from one summary call to the - # next checkpoint. The same fresh-per-attempt, - # generation-scoped discipline the main loop now uses - # (#832); the boundary checks in + # next checkpoint. A fresh generation-scoped ref per + # attempt; the boundary checks in # _summarize_batch/_backoff guarantee a superseded - # compaction makes no further calls — so it can never + # compaction makes no further calls, so it can never # clobber a successor's registration. cancel_ref=_CancelRef(self, my_generation), ) diff --git a/turnstone/core/streaming_text.py b/turnstone/core/streaming_text.py index 333d6688..d0c61351 100644 --- a/turnstone/core/streaming_text.py +++ b/turnstone/core/streaming_text.py @@ -75,22 +75,20 @@ class ThinkTagSplitter: def close_run(self) -> str: """Close the current run at an out-of-band interleave signal. - For the boundary where a provider-parsed ``reasoning_delta`` - arrives mid-stream: everything decided emits at the CURRENT - state, and a possible partial-tag tail - (:func:`partial_tag_tail`) is RETURNED to the caller — the drain - closes its per-run split at the same boundary with the same - rule, which is what keeps the displayed and committed - interpretations of one stream identical there. The carry's - lifetime belongs to the run owner, NOT to :attr:`pending`: the + For a provider-parsed ``reasoning_delta`` arriving mid-stream: + everything decided emits at the CURRENT state, and a possible + partial-tag tail (:func:`partial_tag_tail`) is RETURNED to the + caller. The drain splits its runs at the same boundary by the + same rule, so the displayed and committed readings of one stream + agree. + + The carry belongs to the run owner, not to :attr:`pending`: the tail was cut in the closing run's state, while ``pending`` is read under whatever state later flushes hit (``in_think`` flips across the reasoning block), which would relabel a content-state - carry as reasoning — both halves of the live-caught #832 - display/commit divergence. The caller re-feeds the carry when - content resumes (reassembling a tag the server split across the - block) or flushes it at its original state at a terminal - boundary, mirroring the drain's separate carry variable. + carry as reasoning. The caller re-feeds it when content resumes + — reassembling a tag the server split across the block — or + flushes it at its original state at a terminal boundary. """ if not self.pending: return "" @@ -148,9 +146,9 @@ def partial_tag_tail(text: str) -> str: limit = min(len(text), ThinkTagSplitter.MAX_TAG_LEN - 1) for size in range(limit, 0, -1): suffix = text[-size:] - # Proper prefix only: without the length check a COMPLETE tag - # shorter than the longest one self-matches via startswith and - # gets carried as a "partial", violating the contract above. + # Proper prefix only: without the length check a complete tag + # shorter than the longest one self-matches and gets carried as + # a "partial". if any( len(suffix) < len(tag) and tag.startswith(suffix) for tag in ThinkTagSplitter.ALL_TAGS ):