docs+test(832): retire pre-fold names from prose; pin the eager-append contract

The docs sweep re-points every stale reference to the deleted seam
(architecture.md's flow diagram and ladder inventory, the lowering and
anthropic docstrings, the protocol's shared-rule docstrings that
described the pre-fold dual-assembler world). The Protocol's cancel_ref
contract is strengthened from 'before the first chunk' to 'inside the
call body, before the iterator is returned' — the instant the fold's
creation-vs-midstream classifier and health recording key on — and
three real-SDK-over-mock-transport tripwires pin it per adapter, so a
future lazily-issued generator adapter fails loudly instead of silently
reclassifying every pre-first-chunk death.
This commit is contained in:
Patrick Buckley
2026-08-05 13:58:00 -07:00
parent e103af94e7
commit 5a20916eec
6 changed files with 109 additions and 36 deletions
+16 -11
View File
@@ -128,10 +128,11 @@ A user message flows through the system as follows:
_emit_state("thinking")
|
v
_create_stream_with_retry() ----> provider.create_streaming(client, model, messages, ...)
_stream_response() -------------> model_turn(lane, turns, on_chunk=...) per attempt
| lane-swap fallback walk; per-lane ladder:
| up to 3 retries (4 total attempts), exponential backoff
v
_stream_response(stream) --------> dispatch tokens to UI:
the on_chunk consumer -----------> dispatch tokens to UI:
| on_reasoning_token() / on_content_token()
| accumulate tool_calls from deltas
| track finish_reason
@@ -243,7 +244,9 @@ class SessionUI(Protocol):
def on_content_token(self, text: str) -> None: ...
def on_stream_end(self) -> None: ...
def approve_tools(self, items: list[dict]) -> tuple[bool, str | None]: ...
def on_tool_result(self, call_id: str, name: str, output: str, *, is_error: bool = False) -> None: ...
def on_tool_result(
self, call_id: str, name: str, output: str, *, is_error: bool = False
) -> None: ...
def on_tool_output_chunk(self, call_id: str, chunk: str) -> None: ...
def on_status(self, usage: dict, context_window: int, effort: str) -> None: ...
def on_info(self, message: str) -> None: ...
@@ -333,7 +336,9 @@ class WorkstreamManager:
def __init__(self, session_factory: Callable[[SessionUI], ChatSession]): ...
def create(self, name="", ui_factory=None) -> Workstream: ...
def close(self, ws_id: str) -> bool: ...
def close_idle(self, max_age_seconds: float) -> list[str]: ... # auto-close stale IDLE workstreams
def close_idle(
self, max_age_seconds: float
) -> list[str]: ... # auto-close stale IDLE workstreams
def get(self, ws_id: str) -> Workstream | None: ...
def get_active(self) -> Workstream | None: ...
def list_all(self) -> list[Workstream]: ...
@@ -508,7 +513,7 @@ then returns the final content as the tool result.
response without tools. When unlimited, the loop only exits when the model
stops calling tools or hits `finish_reason: "length"`.
- **Retry**: each API call in the agent loop uses the same retry+backoff logic
as the main `_create_stream_with_retry()`.
as the main loop's per-lane ladder (`_model_turn_with_retry`).
- **Finish reason handling**: `finish_reason: "length"` stops the agent early
and returns whatever content was generated. `finish_reason: "content_filter"`
returns a placeholder.
@@ -941,8 +946,8 @@ with the same alias in-memory (the DB rows are never modified).
5. `/model` command shows available models; `/model <alias>` switches the
active workstream's client, model, context window, and per-model sampling
parameters
6. `_create_stream_with_retry()` tries the primary model, then each fallback
alias in order if the primary is unreachable
6. `_model_turn_with_fallback()` tries the primary lane, then each fallback
alias's lane in order if the primary is unreachable
7. `_run_agent()` resolves `registry.agent_model` (if set) for task
sub-agents, allowing a cheaper model for autonomous loops
@@ -1178,9 +1183,9 @@ Named (aliased) workstreams are never age-pruned. Configure with
Every model call streams (#831); retry lives at two stacked layers:
- **Caller ladders** — `ChatSession._create_stream_with_retry()` (chat
loop) and the agent `_api_call()` (drained via `model_turn`) use the
same pattern: 4 total attempts (1 initial + 3 retries,
- **Caller ladders** — `ChatSession._model_turn_with_retry()` (chat
loop, one ladder per lane) and the agent `_api_call()` (drained via
`model_turn`) use the same pattern: 4 total attempts (1 initial + 3 retries,
`_MAX_RETRIES = 3`), exponential backoff base 1 second
(`delay = 1s * 2^attempt`), `ui.on_info()` on retry, exception
propagates on final failure. `_compact_messages()` wraps its drained
@@ -1274,7 +1279,7 @@ HALF_OPEN ──(probe fails)──────────> OPEN
transition the `_state` (`CircuitState` enum: `CLOSED`, `OPEN`, `HALF_OPEN`).
- `acquire_request_permit()` returns `False` when the circuit is `OPEN` or when
in `HALF_OPEN` and the single probe permit has already been consumed. Causes
`ChatSession._create_stream_with_retry` to skip the backend and surface an
`ChatSession._model_turn_with_fallback` to skip the backend and surface an
error immediately.
- The `/health` endpoint reads the monitor's state: `"status": "ok"` when the
circuit is closed, `"status": "degraded"` when open or half-open.
+59
View File
@@ -207,3 +207,62 @@ def test_cross_thread_client_close_surfaces_transport_error_to_blocked_read():
client.close()
assert not closer_thread.is_alive()
assert not server_thread.is_alive()
class TestEagerAppendContract:
"""Every adapter arms ``cancel_ref`` INSIDE ``create_streaming``'s body —
at HTTP-response time, before the iterator is returned (the Protocol
contract, strengthened on #832): the interactive wrapper's
creation-vs-midstream classifier and its health recording key on that
instant, and a lazily-issued generator adapter would silently move the
arming to first ``next()``, misclassifying every pre-first-chunk death
as a creation failure. Real SDK clients over mock transports; the
assertion deliberately runs BEFORE any iteration.
"""
def _armed_at_return(self, provider, client, **extra):
ref: list = []
stream = provider.create_streaming(
client=client,
model="m",
messages=[{"role": "user", "content": "hi"}],
cancel_ref=ref,
**extra,
)
assert len(ref) == 1, "cancel_ref not armed before create_streaming returned"
assert hasattr(ref[0], "close")
with contextlib.suppress(Exception):
stream.close()
def test_openai_chat_arms_eagerly(self):
from turnstone.core.providers._openai_chat import OpenAIChatCompletionsProvider
requests: list = []
client = openai.OpenAI(
api_key="probe",
http_client=httpx.Client(transport=_dying_transport(CHAT_CHUNK, requests)),
)
self._armed_at_return(OpenAIChatCompletionsProvider(), client)
assert len(requests) == 1 # the HTTP call happened inside create
def test_openai_responses_arms_eagerly(self):
from turnstone.core.providers._openai_responses import OpenAIResponsesProvider
requests: list = []
client = openai.OpenAI(
api_key="probe",
http_client=httpx.Client(transport=_dying_transport(CHAT_CHUNK, requests)),
)
self._armed_at_return(OpenAIResponsesProvider(), client)
assert len(requests) == 1
def test_anthropic_arms_eagerly(self):
from turnstone.core.providers._anthropic import AnthropicProvider
requests: list = []
client = anthropic.Anthropic(
api_key="probe",
http_client=httpx.Client(transport=_dying_transport(ANTHROPIC_EVENTS, requests)),
)
self._armed_at_return(AnthropicProvider(), client)
assert len(requests) == 1
+1 -1
View File
@@ -284,7 +284,7 @@ def sanitize_tool_call_arguments(messages: list[dict[str, Any]]) -> list[dict[st
The stream accumulator commits ``arguments`` verbatim, and the only guard that
drops a malformed tool call is ``finish_reason == "length"``
(``ChatSession._stream_attempt``); a model that emits invalid JSON with a
(``ChatSession._finalize_stream_result``); a model that emits invalid JSON with a
``stop`` / ``tool_calls`` finish reason slips through, and one such turn then
poison-pills every later request that replays it on a strict renderer. This
legalizes each offending ``arguments`` to a JSON-object string.
+5 -5
View File
@@ -554,8 +554,8 @@ class AnthropicProvider:
intentionally preserved. The kwarg defaults to ``True`` here
purely for back-compat with any direct caller that hasn't been
updated to thread the resolver — production call sites
(``ChatSession._try_stream`` / ``_utility_completion``) always
pass the resolved flag explicitly.
(``model_turn`` — every lane, the interactive loop included)
always pass the resolved flag explicitly.
``supports_mid_conversation_system`` (claude-opus-4-8,
claude-fable-5) makes the
@@ -1306,9 +1306,9 @@ def _normalize_finish_reason(reason: str) -> str:
# drain gate only errors on an ABSENT finish reason. "content_filter"
# is the
# OpenAI-vocabulary equivalent this function normalizes onto, and both
# consumers already handle it: ChatSession._stream_attempt warns the
# user, and the sub-agent loop stops early instead of flailing on an
# empty turn. Falling through to the raw "refusal" string instead
# consumers already handle it: the interactive wrapper's
# _finalize_stream_result warns the user, and the sub-agent loop
# stops early instead of flailing on an empty turn. Falling through to the raw "refusal" string instead
# would land a truncated answer as a complete result.
return "content_filter"
return reason
+15 -8
View File
@@ -98,10 +98,11 @@ def merge_usage(acc: UsageInfo | None, new: UsageInfo) -> UsageInfo:
recomputed from the merged parts. Returns a fresh ``UsageInfo`` and
never mutates ``new`` (the provider's object).
Serves :func:`drain_stream` and ``ChatSession``'s inline chunk
consumer alike — the interactive loop re-projects the merged
accumulator into its ``_last_usage`` dict on every usage chunk (that
dict has mid-stream readers), so the two lanes cannot drift on the
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.
"""
if acc is None:
@@ -146,8 +147,9 @@ 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`, ``GoogleProvider``'s raw-fidelity
capture, and ``ChatSession``'s inline chunk consumer — every
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.
"""
@@ -874,8 +876,13 @@ class LLMProvider(Protocol):
are respected.
If *cancel_ref* is provided the provider appends the underlying SDK
stream object (which has a ``.close()`` method) before yielding the
first chunk. The caller can then close it from another thread to
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.
This is the ONLY transport — single-shot callers drain it through
+6 -4
View File
@@ -525,7 +525,9 @@ class _StreamTurnConsumer:
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
stale-usage leak (the old ``_stream_attempt``-entry comment) and
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.
"""
s = self._session
@@ -3046,7 +3048,7 @@ class ChatSession:
Thread safety: each assignment creates a new object (copy-on-write).
Under CPython's GIL, individual reference assignments are atomic.
``_try_stream`` captures tools at call time, so a concurrent refresh
Each attempt captures tools at call time, so a concurrent refresh
between turns is safe; mid-stream the LLM request already holds
the old snapshot.
"""
@@ -4941,7 +4943,7 @@ class ChatSession:
fires on a history render."""
if not ids:
return {}
# caps is the ACTIVE attempt's capabilities, threaded from _try_stream so
# caps is the ACTIVE attempt's capabilities, threaded from its lane so
# a fallback to a model with different media support converts on the
# right caps; default to the primary only when called without one.
if caps is None:
@@ -7787,7 +7789,7 @@ class ChatSession:
# flagged ("…cannot simultaneously guarantee Consistency,"
# surfaced as if it were a complete sentence).
if self._cancelled_partial_msg:
# _stream_attempt was interrupted — save partial
# the streaming attempt was interrupted — save partial
# assistant msg. Two shapes:
#
# - Some text streamed before cancel: append the