feat(session): fold the main streaming loop onto model_turn (#832)

The send path's plant call is now one model_turn invocation per attempt,
reached through a lane-swap fallback walk that mirrors the old creation
ladder 1:1: an inner per-lane retry (_model_turn_with_retry) inside the
two-pass healthy/degraded walk (_model_turn_with_fallback), with health
success recorded at the request-accepted instant via the per-attempt
_CancelRef's new on_first_append hook and failure once per lane ladder.
The hook is also the creation-vs-midstream classifier: an armed attempt's
death re-raises to the re-issue ladder on every lane — a fallback stream
that died after tokens reached the UI is never swallowed into
try-the-next-alias — and carries the per-turn usage-slot resets at the
old timing so a reconnecting tab's status bar never blanks mid-walk.

Chunk-to-UI translation lives in _StreamTurnConsumer (model_turn's
on_chunk body): display-side only, the canonical turn always assembled by
drain_stream at the one seam; the inline-tag scan reads the SAME lane
capability the drain gate reads (server_parses_reasoning), replacing the
creation-time handoff register — which is deleted — so display and commit
cannot disagree about a backend's posture, fallback walk included.
Cancellation converges: every model-call site now builds fresh
generation-scoped refs, closing the force-cancel hole where the old gen-0
shared ref read aborted=False for an orphaned generation and would have
let a retry re-issue on its behalf; the pre-dispatch abort read inside
model_turn also means a Stop set before the turn no longer mints a
credential on a dynamically authenticated alias.

send() consumes the result natively: the committed Turn carries minted
ids, the finalized native lane, and an accurate producer — fixing the
latent mislabel where fallback-served turns were persisted under the
primary provider's name, and the fork asymmetry where in-memory turns
decoded with producer="". Ruled behavior changes (design D12): the
trailing citations footer now folds into committed content (it previously
lived only in an ephemeral info bubble and vanished on reload); a stream
that exhausts without a finish reason is a retryable mid-stream death
instead of a silent partial commit; length-truncated turns keep dropping
partial tool calls, now as an explicit post-drain policy. The replay
parity harness pins all thirteen scenarios against pre-fold baselines,
transformed only where a ruling applies — and caught two real bugs during
the fold (the splitter's end-of-stream carry never flushing to the UI,
and the footer splicing into the answer's held tail).

ChatSession imports no provider module: create_streaming has exactly one
caller module, and the protocol types, merge_usage, and create_provider
reach the session through model_turn's re-export seam.

tests/_parity_832.py

@@ -40,6 +40,7 @@ from turnstone.core.providers._protocol import (
     ToolCallDelta,
     UsageInfo,
 )
+from turnstone.core.trajectory import Turn
 
 FIXTURE_DIR = Path(__file__).parent / "data" / "parity_832"
 UPDATE = os.environ.get("UPDATE_832_PARITY") == "1"
@@ -198,16 +199,17 @@ def run_scenario(name: str) -> dict[str, Any]:
     ui = RecordingUI()
     session = make_session(ui=ui)
     session._provider = scripted_provider(SCENARIOS[name])
-    msgs = [{"role": "user", "content": "hi"}]
+    session.messages.append(Turn.user("hi"))
 
     record: dict[str, Any] = {"scenario": name}
     try:
-        msg = session._stream_response(msgs, 0)
-        msg.pop("_wire_msgs", None)
+        result = session._stream_response(0)
         record["result"] = {
-            "content": msg.get("content", ""),
-            "tool_calls": msg.get("tool_calls"),
-            "provider_content": msg.get("_provider_content"),
+            "content": result.content,
+            "tool_calls": result.tool_calls or None,
+            "provider_content": (
+                [dict(b) for b in result.turn.native.blocks] if result.turn.native else None
+            ),
         }
         record["raised"] = None
     except BaseException as exc:  # noqa: BLE001 — the record IS the observation

tests/test_832_parity.py

@@ -35,10 +35,57 @@ from tests._parity_832 import (
 def _apply_ruled_deltas(name: str, baseline: dict[str, Any]) -> dict[str, Any]:
     """Transform an old-world record into the post-fold expectation.
 
-    Every transform cites its D12 row.  Pre-fold trees (fixtures being
-    captured) never reach this — capture mode writes and exits.
+    Every transform cites its D12 row (docs/design/832-main-loop-model-turn.md).
+    Pre-fold trees (fixtures being captured) never reach this — capture mode
+    writes and exits.
     """
     expected = json.loads(json.dumps(baseline))  # deep copy
+
+    if name == "info_postfinish_footer":
+        # D12 row 1 (RULED adopt-drain): the trailing citations footer
+        # enters the COMMITTED content (conditional fold: non-blank answer,
+        # "\n\n" separator) and streams as content — post-carry-flush, so
+        # displayed ordering matches committed — instead of an ephemeral
+        # info bubble that never survived reload.
+        footer = "Sources:\n- example.com/page"
+        expected["result"]["content"] += "\n\n" + footer
+        events = [e for e in expected["ui_events"] if e != ["info", footer]]
+        end = events.index(["stream_end", ""])
+        events[end:end] = [["content", "\n\n" + footer]]
+        expected["ui_events"] = events
+
+    elif name == "no_finish_clean_exhaust":
+        # D12 row 2 (RULED adopt strict gate): a stream that exhausts with
+        # no finish reason no 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"
+        expected["result"] = None
+        retry_theater = []
+        for attempt in (1, 2):
+            retry_theater += [
+                ["stream_end", ""],
+                [
+                    "info",
+                    f"[stream died mid-response (IncompleteStreamError) — retrying in "
+                    f"{2 ** (attempt - 1)}s ({attempt}/2)]",
+                ],
+                ["stream_discarded", ""],
+                ["thinking_start", ""],
+                ["thinking_stop", ""],
+            ]
+        expected["ui_events"] = (
+            [["thinking_stop", ""]] + retry_theater + [["stream_end", ""], ["stream_discarded", ""]]
+        )
+
+    elif name == "think_tags_split_across_chunks":
+        # D12 residue-trim row (RULED adopt; V14.1): the COMMITTED content
+        # takes the drain's single edge trim when a tag was consumed; the
+        # displayed stream keeps the raw residue ("\n\nAnswer") — the
+        # accepted snapshot-vs-history whitespace class.
+        expected["result"]["content"] = expected["result"]["content"].lstrip("\n")
+
     return expected
 
 

turnstone/core/model_turn.py

@@ -59,6 +59,12 @@ from turnstone.core.lowering import (
     sanitize_tool_call_arguments,
 )
 
+# The provider FACTORY rides the same re-export seam: the session's
+# no-registry default construction is the one provider-construction site
+# left outside the registry, and routing it through here keeps the
+# provider package a plant-layer-only import.
+from turnstone.core.providers import create_provider as create_provider
+
 # Protocol names imported at runtime (not TYPE_CHECKING) and re-exported with
 # the explicit ``as`` idiom: post-#832 ``ChatSession`` types its provider
 # handles, chunk callback, and capabilities against THIS module, so the
@@ -759,6 +765,12 @@ class ModelTurnResult:
     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).
+
+    *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).
     """
 
     turn: Turn
@@ -766,6 +778,7 @@ class ModelTurnResult:
     usage: UsageInfo | None
     tool_calls: list[dict[str, Any]]
     wire_msgs: list[dict[str, Any]] | None = None
+    producer: str = ""
 
     @property
     def content(self) -> str:
@@ -1144,4 +1157,5 @@ def model_turn(
         usage=result.usage,
         tool_calls=raw_calls,
         wire_msgs=wire,
+        producer=lane.provider.provider_name,
     )