diff --git a/docs/architecture.md b/docs/architecture.md index 382d35c8..a911d189 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -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: ... @@ -313,15 +316,15 @@ ERROR last operation failed ```python @dataclass class Workstream: - id: str # uuid hex, 8 chars - name: str # user-visible label - state: WorkstreamState # current state - session: ChatSession | None # the conversation engine - ui: SessionUI | None # frontend adapter + id: str # uuid hex, 8 chars + name: str # user-visible label + state: WorkstreamState # current state + session: ChatSession | None # the conversation engine + ui: SessionUI | None # frontend adapter worker_thread: threading.Thread | None error_message: str - last_active: float # time.monotonic() timestamp, updated on every state change - _lock: threading.Lock # per-workstream state lock + last_active: float # time.monotonic() timestamp, updated on every state change + _lock: threading.Lock # per-workstream state lock ``` ### WorkstreamManager @@ -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 ` 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. diff --git a/tests/test_sdk_stream_boundary.py b/tests/test_sdk_stream_boundary.py index b690d3bb..2176dcb6 100644 --- a/tests/test_sdk_stream_boundary.py +++ b/tests/test_sdk_stream_boundary.py @@ -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 diff --git a/turnstone/core/lowering.py b/turnstone/core/lowering.py index 0407b38f..2fad5707 100644 --- a/turnstone/core/lowering.py +++ b/turnstone/core/lowering.py @@ -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. diff --git a/turnstone/core/providers/_anthropic.py b/turnstone/core/providers/_anthropic.py index ff1a3b9d..e92607f6 100644 --- a/turnstone/core/providers/_anthropic.py +++ b/turnstone/core/providers/_anthropic.py @@ -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 diff --git a/turnstone/core/providers/_protocol.py b/turnstone/core/providers/_protocol.py index 2ef563d7..2d80ecdb 100644 --- a/turnstone/core/providers/_protocol.py +++ b/turnstone/core/providers/_protocol.py @@ -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 diff --git a/turnstone/core/session.py b/turnstone/core/session.py index 4d6f8eb8..73686908 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -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