test(parity): strengthen core streaming contracts

tests/route_parity/README.md

@@ -21,6 +21,30 @@
 - Every test saves and restores the original bridge state
 - A small subprocess smoke test verifies environment-based startup configuration and detects fallback to the Python HTTP implementation
 
+## Streaming execution
+
+The invocation callback passed to `run_in_process` must consume the stream before returning its `StreamOutcome`.
+Use `consume_sync_stream` inside that callback, or await `consume_async_stream` inside the callback passed to
+`run_in_process_async`. Provider requests are collected only after the callback completes. Streaming is explicit:
+an iterable return value alone does not select stream consumption
+
+The consumers retain the wrapper type, iteration capabilities, chunk types and order, and any partial output before
+an error. Errors retain their creation or iteration phase and the full public `SDKError` fields, with traceback text
+removed. `capture_sync_stream` and `capture_async_stream` consume through the same helpers and then serialize the
+outcome for subprocess reports. A serialization failure raises as a harness failure rather than becoming an SDK error
+
+Response models and stream chunks share a recursive comparator. It compares concrete model, container, and scalar
+types, public fields and extras, and exact values while ignoring Pydantic private attributes at every nesting level.
+An API may supply an explicit chunk normalizer for its public contract
+
+Shared tests exercise a local SSE provider through recording, VCR cassette storage, replay, and typed event comparison
+in sync and async modes. They cover fragmented events, split UTF-8 characters, CRLF framing, coalesced events, and
+application errors within a normally completed HTTP stream. HTTP byte boundaries and decoded SDK event boundaries
+are checked separately
+
+OCR remains the only integrated LiteLLM route. These tests validate shared streaming machinery, not another route's
+SDK parity. Connection interruption, early cancellation, and lifecycle timeout enforcement remain outside this coverage
+
 ## Hypothesis and property-based testing
 
 - Hypothesis is a Python library for property-based testing

tests/route_parity/compare.py

@@ -1,14 +1,12 @@
 from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
-from typing import Final, TypeVar, cast
+from typing import Final, cast
 
 from pydantic import BaseModel
 
 from tests.route_parity.models import CapturedRequest, Execution
 
-ModelT = TypeVar("ModelT", bound=BaseModel)
-
 
 def validate_harness(baseline: Execution, candidate: Execution, baseline_user_agent: str) -> None:
     for request in baseline.requests:
@@ -29,31 +27,35 @@ def _request_after_transformation(request: CapturedRequest) -> CapturedRequest:
 def assert_request_parity(baseline: tuple[CapturedRequest, ...], candidate: tuple[CapturedRequest, ...]) -> None:
     baseline_requests: Final = tuple(_request_after_transformation(request) for request in baseline)
     candidate_requests: Final = tuple(_request_after_transformation(request) for request in candidate)
-    assert baseline_requests == candidate_requests
+    assert_value_parity(baseline_requests, candidate_requests)
 
 
-def public_model_copy(model: ModelT) -> ModelT:
-    copied: Final = model.model_copy(deep=True)
-    object.__setattr__(copied, "__pydantic_private__", None)
-    return copied
+def _public_model_values(model: BaseModel) -> dict[str, object]:
+    fields: Final = (*type(model).model_fields, *type(model).model_computed_fields)
+    extras: Final = cast(Mapping[str, object], model.model_extra or {})
+    return {
+        **{name: cast(object, getattr(model, name)) for name in fields if not name.startswith("_")},
+        **{name: value for name, value in extras.items() if not name.startswith("_")},
+    }
 
 
 def assert_model_parity(baseline: BaseModel, candidate: BaseModel) -> None:
-    assert type(baseline) is type(candidate)
-    _assert_value_parity(
-        public_model_copy(baseline).model_dump(mode="python"),
-        public_model_copy(candidate).model_dump(mode="python"),
-        path="$",
-    )
+    assert_value_parity(baseline, candidate)
 
 
-def _assert_value_parity(baseline: object, candidate: object, *, path: str) -> None:
+def assert_value_parity(baseline: object, candidate: object, *, path: str = "$") -> None:
+    assert type(baseline) is type(candidate), f"type mismatch at {path}: {type(baseline)} != {type(candidate)}"
+    if isinstance(baseline, BaseModel) and isinstance(candidate, BaseModel):
+        assert_value_parity(_public_model_values(baseline), _public_model_values(candidate), path=path)
+        return
     if isinstance(baseline, Mapping) and isinstance(candidate, Mapping):
         baseline_mapping: Final = cast(Mapping[object, object], baseline)
         candidate_mapping: Final = cast(Mapping[object, object], candidate)
-        assert baseline_mapping.keys() == candidate_mapping.keys(), f"mapping keys differ at {path}"
+        assert frozenset((type(key), key) for key in baseline_mapping) == frozenset(
+            (type(key), key) for key in candidate_mapping
+        ), f"mapping keys differ at {path}"
         for key in baseline_mapping:
-            _assert_value_parity(baseline_mapping[key], candidate_mapping[key], path=f"{path}.{key}")
+            assert_value_parity(baseline_mapping[key], candidate_mapping[key], path=f"{path}.{key}")
         return
     if (
         isinstance(baseline, Sequence)
@@ -64,8 +66,10 @@ def _assert_value_parity(baseline: object, candidate: object, *, path: str) -> N
         baseline_sequence: Final = cast(Sequence[object], baseline)
         candidate_sequence: Final = cast(Sequence[object], candidate)
         assert len(baseline_sequence) == len(candidate_sequence), f"sequence lengths differ at {path}"
-        for index, (baseline_item, candidate_item) in enumerate(zip(baseline_sequence, candidate_sequence)):
-            _assert_value_parity(baseline_item, candidate_item, path=f"{path}[{index}]")
+        for index, (baseline_item, candidate_item) in enumerate(
+            zip(baseline_sequence, candidate_sequence, strict=True)
+        ):
+            assert_value_parity(baseline_item, candidate_item, path=f"{path}[{index}]")
         return
     assert baseline == candidate, f"value mismatch at {path}: {baseline!r} != {candidate!r}"
 
@@ -73,4 +77,4 @@ def _assert_value_parity(baseline: object, candidate: object, *, path: str) -> N
 def assert_parity(baseline: Execution, candidate: Execution, baseline_user_agent: str) -> None:
     validate_harness(baseline, candidate, baseline_user_agent)
     assert_request_parity(baseline.requests, candidate.requests)
-    assert baseline.report == candidate.report
+    assert_value_parity(baseline.report, candidate.report)

tests/route_parity/fixtures/test_recording.py

@@ -1,18 +1,22 @@
 from __future__ import annotations
 
+import asyncio
+import queue
 import threading
-from collections.abc import Callable, Generator
+from collections.abc import AsyncIterator, Callable, Generator, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
-from typing import Final
+from typing import Final, Literal
 
 import httpx
 import pytest
 from hypothesis import strategies as st
+from openai._streaming import SSEDecoder
 from pydantic import BaseModel, ConfigDict
 
+from tests.route_parity.compare import assert_request_parity
 from tests.route_parity.fixtures.pipeline import RecordingTarget, record_fixtures
 from tests.route_parity.fixtures.recording import (
     UpstreamEndpoint,
@@ -22,15 +26,25 @@ from tests.route_parity.fixtures.recording import (
 from tests.route_parity.fixtures.store import (
     FIXTURE_SCHEMA_VERSION,
     fixture_path,
+    load_fixture,
     recorded_fixtures,
 )
+from tests.route_parity.inprocess import InProcessExecution, run_in_process, run_in_process_async
 from tests.route_parity.recorded_http import (
     HttpHeader,
     RecordedHttpStreamResponse,
     RecordedResponse,
     RecordedStreamChunk,
 )
-from tests.route_parity.replay import replay_server
+from tests.route_parity.replay import ReplayServer, replay_server
+from tests.route_parity.stream import (
+    StreamCompleted,
+    StreamFailed,
+    StreamOutcome,
+    assert_stream_parity,
+    consume_async_stream,
+    consume_sync_stream,
+)
 
 _SSE_CHUNKS: Final = (
     b'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n',
@@ -39,6 +53,75 @@ _SSE_CHUNKS: Final = (
 )
 
 
+class _StreamEvent(BaseModel):
+    kind: Literal["delta", "done", "error"]
+    value: str
+
+
+class _StreamApplicationError(Exception):
+    status_code: Final = 400
+    code: Final = "invalid_input"
+    type: Final = "validation_error"
+    param: Final = "input"
+    model: Final = "fixture-model"
+    llm_provider: Final = "fixture-provider"
+
+
+def _stream_event(data: str) -> _StreamEvent:
+    event: Final = _StreamEvent.model_validate_json(data)
+    if event.kind == "error":
+        raise _StreamApplicationError(event.value)
+    return event
+
+
+def _event_chunks(failed: bool) -> tuple[bytes, ...]:
+    terminal: Final = (
+        b'event: error\r\ndata: {"kind":"error","value":"invalid input"}\r\n\r\n'
+        if failed
+        else b'event: done\r\ndata: {"kind":"done","value":""}\r\n\r\n'
+    )
+    return (
+        b'event: delta\r\ndata: {"kind":"delta",\r\ndata: "value":"caf\xc3',
+        b'\xa9"}\r\n',
+        b'\r\nevent: delta\r\ndata: {"kind":"delta","value":"second"}\r\n\r\n' + terminal,
+    )
+
+
+def _sync_events(api_base: str, case_input: _FixtureInput) -> Iterator[_StreamEvent]:
+    with httpx.stream("POST", f"{api_base}/stream", json={"id": case_input.identifier}, timeout=5) as response:
+        response.raise_for_status()
+        for event in SSEDecoder().iter_bytes(response.iter_bytes()):
+            yield _stream_event(event.data)
+
+
+async def _async_events(api_base: str, case_input: _FixtureInput) -> AsyncIterator[_StreamEvent]:
+    async with httpx.AsyncClient(timeout=5) as client:
+        async with client.stream("POST", f"{api_base}/stream", json={"id": case_input.identifier}) as response:
+            response.raise_for_status()
+            async for event in SSEDecoder().aiter_bytes(response.aiter_bytes()):
+                yield _stream_event(event.data)
+
+
+async def _consume_async_events(api_base: str, case_input: _FixtureInput) -> StreamOutcome:
+    async def create() -> AsyncIterator[_StreamEvent]:
+        return _async_events(api_base, case_input)
+
+    return await consume_async_stream(create)
+
+
+async def _replay_events(
+    mode: Literal["sync", "async"],
+    provider: ReplayServer,
+    response: RecordedHttpStreamResponse,
+    case_input: _FixtureInput,
+) -> InProcessExecution[StreamOutcome]:
+    if mode == "sync":
+        return run_in_process(
+            provider, (response,), lambda url: consume_sync_stream(lambda: _sync_events(url, case_input))
+        )
+    return await run_in_process_async(provider, (response,), lambda url: _consume_async_events(url, case_input))
+
+
 class _FixtureInput(BaseModel):
     model_config = ConfigDict(frozen=True, extra="forbid")
 
@@ -66,8 +149,9 @@ class _Invocation:
 class _ControlledUpstream(ThreadingHTTPServer):
     daemon_threads = True
 
-    def __init__(self) -> None:
+    def __init__(self, stream_chunks: tuple[bytes, ...]) -> None:
         super().__init__(("127.0.0.1", 0), _ControlledUpstreamHandler)
+        self.stream_chunks: Final = stream_chunks
         self.lock: Final = threading.Lock()
         self.two_requests_started: Final = threading.Event()
         self.active_requests: int = 0
@@ -116,14 +200,14 @@ class _ControlledUpstreamHandler(BaseHTTPRequestHandler):
             self.send_header("content-length", "0")
             self.end_headers()
             return
-        if self.path == "/v1/chat/completions":
+        if self.path in {"/v1/chat/completions", "/stream"}:
             with upstream.lock:
                 upstream.request_count += 1
             self.send_response(200)
             self.send_header("content-type", "text/event-stream")
             self.send_header("transfer-encoding", "chunked")
             self.end_headers()
-            for chunk in _SSE_CHUNKS:
+            for chunk in upstream.stream_chunks:
                 self.wfile.write(f"{len(chunk):X}\r\n".encode("ascii"))
                 self.wfile.write(chunk)
                 self.wfile.write(b"\r\n")
@@ -173,8 +257,8 @@ class _ControlledUpstreamHandler(BaseHTTPRequestHandler):
 
 
 @contextmanager
-def _controlled_upstream() -> Generator[_ControlledUpstream]:
-    server: Final = _ControlledUpstream()
+def _controlled_upstream(stream_chunks: tuple[bytes, ...] = _SSE_CHUNKS) -> Generator[_ControlledUpstream]:
+    server: Final = _ControlledUpstream(stream_chunks)
     thread: Final = threading.Thread(target=server.serve_forever, daemon=True)
     thread.start()
     try:
@@ -430,3 +514,61 @@ def test_multiple_provider_calls_record_and_replay_in_order(
         requests: Final = provider.take_requests(2)
 
     assert len(requests) == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ("sync", "async"))
+@pytest.mark.parametrize("failed", (False, True), ids=("completed", "application-error"))
+async def test_typed_stream_recording_cassette_replay_parity(
+    tmp_path: Path, mode: Literal["sync", "async"], failed: bool
+) -> None:
+    case_input: Final = _case("typed-stream")
+    outcomes: Final[queue.SimpleQueue[StreamOutcome]] = queue.SimpleQueue()
+
+    def record(api_base: str, sdk_input: _FixtureInput) -> None:
+        outcome: Final = (
+            consume_sync_stream(lambda: _sync_events(api_base, sdk_input))
+            if mode == "sync"
+            else asyncio.run(_consume_async_events(api_base, sdk_input))
+        )
+        outcomes.put(outcome)
+
+    with _controlled_upstream(_event_chunks(failed)) as upstream:
+        target: Final = RecordingTarget(
+            name="stream",
+            upstream=UpstreamEndpoint(upstream.url),
+            strategy=st.just(case_input),
+            invocation=_Invocation(record),
+        )
+        summary: Final = record_fixtures((target,), tmp_path, 1, 1, _ParityCase)
+
+    assert summary.failed == ()
+    assert len(summary.recorded) == 1
+    recorded: Final = outcomes.get_nowait()
+    loaded: Final = load_fixture(tmp_path / "stream", case_input, _ParityCase)
+    assert loaded is not None
+    response: Final = loaded.provider_responses[0]
+    assert isinstance(response, RecordedHttpStreamResponse)
+    assert response.status_code == 200
+    wire_bytes: Final = b"".join(chunk.data_bytes() for chunk in response.chunks)
+    assert wire_bytes == b"".join(_event_chunks(failed))
+    coalesced: Final = response.model_copy(update={"chunks": (RecordedStreamChunk.from_bytes(wire_bytes),)})
+
+    with replay_server() as provider:
+        first: Final = await _replay_events(mode, provider, response, case_input)
+        second: Final = await _replay_events(mode, provider, coalesced, case_input)
+    assert_request_parity(first.requests, second.requests)
+    assert len(first.requests) == 1
+    assert first.requests[0].body == {"id": case_input.identifier}
+    assert_stream_parity(recorded, first.response)
+    assert_stream_parity(first.response, second.response)
+    expected: Final = (_StreamEvent(kind="delta", value="café"), _StreamEvent(kind="delta", value="second"))
+    assert first.response.chunks == (expected if failed else (*expected, _StreamEvent(kind="done", value="")))
+    if failed:
+        assert isinstance(first.response.terminal, StreamFailed)
+        assert first.response.terminal.phase == "iteration"
+        assert first.response.terminal.exception_type is _StreamApplicationError
+        assert first.response.terminal.error.code == "invalid_input"
+        assert first.response.terminal.error.message == "invalid input"
+    else:
+        assert first.response.terminal == StreamCompleted()

tests/route_parity/inprocess.py

@@ -1,8 +1,8 @@
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
-from typing import Generic, TypeVar
+from typing import Final, Generic, TypeVar
 
 from tests.route_parity.models import CapturedRequest
 from tests.route_parity.recorded_http import RecordedResponse
@@ -25,7 +25,22 @@ def run_in_process(
     for recorded_response in recorded_responses:
         provider.enqueue_response(recorded_response)
     try:
-        response = call(provider.url)
+        response: Final = call(provider.url)
+        return InProcessExecution(requests=provider.take_requests(len(recorded_responses)), response=response)
+    except Exception:
+        provider.reset()
+        raise
+
+
+async def run_in_process_async(
+    provider: ReplayServer,
+    recorded_responses: tuple[RecordedResponse, ...],
+    call: Callable[[str], Awaitable[ResponseT]],
+) -> InProcessExecution[ResponseT]:
+    for recorded_response in recorded_responses:
+        provider.enqueue_response(recorded_response)
+    try:
+        response: Final = await call(provider.url)
         return InProcessExecution(requests=provider.take_requests(len(recorded_responses)), response=response)
     except Exception:
         provider.reset()

tests/route_parity/stream.py

@@ -4,11 +4,9 @@ from collections.abc import AsyncIterable, Awaitable, Callable, Iterable
 from dataclasses import dataclass
 from typing import Final, Literal, TypeAlias
 
-from pydantic import BaseModel
-
-from tests.route_parity.compare import public_model_copy
+from tests.route_parity.compare import assert_value_parity
 from tests.route_parity.models import (
-    SDKChunk,
+    SDKError,
     SDKReport,
     SDKStreamCompleted,
     SDKStreamFailed,
@@ -27,9 +25,7 @@ class StreamCompleted:
 class StreamFailed:
     phase: Literal["creation", "iteration"]
     exception_type: type[BaseException]
-    status_code: int | None
-    llm_provider: str | None
-    model: str | None
+    error: SDKError
     kind: Literal["failed"] = "failed"
 
 
@@ -60,51 +56,34 @@ async def drain_async_stream(stream: AsyncIterable[object]) -> None:
 
 
 def capture_sync_stream(create: Callable[[], Iterable[object]]) -> SDKReport:
-    try:
-        stream: Final = create()
-    except Exception as error:
-        return sdk_error_report(error)
-
-    chunks: list[SDKChunk] = []  # mutable-ok: partial chunks must survive an iteration failure
-    try:
-        for chunk in stream:
-            chunks.append(sdk_chunk(chunk))
-    except Exception as error:
-        return SDKStreamReport(chunks=tuple(chunks), terminal=SDKStreamFailed(error=sdk_error_report(error)))
-    return SDKStreamReport(chunks=tuple(chunks), terminal=SDKStreamCompleted())
+    return _stream_report(consume_sync_stream(create))
 
 
 async def capture_async_stream(create: Callable[[], Awaitable[AsyncIterable[object]]]) -> SDKReport:
-    try:
-        stream: Final = await create()
-    except Exception as error:
-        return sdk_error_report(error)
-
-    chunks: list[SDKChunk] = []  # mutable-ok: partial chunks must survive an iteration failure
-    try:
-        async for chunk in stream:
-            chunks.append(sdk_chunk(chunk))
-    except Exception as error:
-        return SDKStreamReport(chunks=tuple(chunks), terminal=SDKStreamFailed(error=sdk_error_report(error)))
-    return SDKStreamReport(chunks=tuple(chunks), terminal=SDKStreamCompleted())
+    return _stream_report(await consume_async_stream(create))
 
 
-def _failed(phase: Literal["creation", "iteration"], error: BaseException) -> StreamFailed:
-    status_code: Final = getattr(error, "status_code", None)
-    llm_provider: Final = getattr(error, "llm_provider", None)
-    model: Final = getattr(error, "model", None)
+def _stream_report(outcome: StreamOutcome) -> SDKReport:
+    terminal: Final = outcome.terminal
+    if isinstance(terminal, StreamFailed) and terminal.phase == "creation":
+        return terminal.error
+    return SDKStreamReport(
+        chunks=tuple(sdk_chunk(chunk) for chunk in outcome.chunks),
+        terminal=SDKStreamFailed(error=terminal.error) if isinstance(terminal, StreamFailed) else SDKStreamCompleted(),
+    )
+
+
+def _failed(phase: Literal["creation", "iteration"], error: Exception) -> StreamFailed:
     return StreamFailed(
         phase=phase,
         exception_type=type(error),
-        status_code=status_code if isinstance(status_code, int) else None,
-        llm_provider=llm_provider if isinstance(llm_provider, str) else None,
-        model=model if isinstance(model, str) else None,
+        error=sdk_error_report(error),
     )
 
 
 def consume_sync_stream(create: Callable[[], Iterable[object]]) -> StreamOutcome:
     try:
-        stream = create()
+        stream: Final = create()
     except Exception as error:
         return StreamOutcome(
             wrapper_type=None,
@@ -142,7 +121,7 @@ def consume_sync_stream(create: Callable[[], Iterable[object]]) -> StreamOutcome
 
 async def consume_async_stream(create: Callable[[], Awaitable[AsyncIterable[object]]]) -> StreamOutcome:
     try:
-        stream = await create()
+        stream: Final = await create()
     except Exception as error:
         return StreamOutcome(
             wrapper_type=None,
@@ -179,8 +158,6 @@ async def consume_async_stream(create: Callable[[], Awaitable[AsyncIterable[obje
 
 
 def normalize_chunk(chunk: object) -> object:
-    if isinstance(chunk, BaseModel):
-        return public_model_copy(chunk)
     return chunk
 
 
@@ -195,6 +172,6 @@ def assert_stream_parity(
     assert baseline.supports_async_iteration is candidate.supports_async_iteration
     assert baseline.chunk_types == candidate.chunk_types
     assert len(baseline.chunks) == len(candidate.chunks)
-    for baseline_chunk, candidate_chunk in zip(baseline.chunks, candidate.chunks, strict=True):
-        assert normalize(baseline_chunk) == normalize(candidate_chunk)
+    for index, (baseline_chunk, candidate_chunk) in enumerate(zip(baseline.chunks, candidate.chunks, strict=True)):
+        assert_value_parity(normalize(baseline_chunk), normalize(candidate_chunk), path=f"$.chunks[{index}]")
     assert baseline.terminal == candidate.terminal

tests/route_parity/test_parity.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
+from types import MappingProxyType
 from typing import Final
 
 import pytest
-from pydantic import BaseModel, JsonValue, PrivateAttr
+from pydantic import BaseModel, ConfigDict, JsonValue, PrivateAttr
 
 from tests.route_parity.compare import assert_model_parity, assert_parity
 from tests.route_parity.models import CapturedRequest, Execution, SDKError, SDKSuccess, sdk_error_report
@@ -27,6 +28,12 @@ class _FloatResponse(BaseModel):
     values: list[float]
 
 
+class _PublicValue(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    value: object
+
+
 class _PublicError(ValueError):
     status_code: Final = 400
 
@@ -136,3 +143,48 @@ def test_model_parity_rejects_meaningful_float_difference() -> None:
             _FloatResponse(values=[0.22590550796036835]),
             _FloatResponse(values=[0.2259]),
         )
+
+
+@pytest.mark.parametrize(
+    ("baseline", "candidate"),
+    (
+        (_ComparableResponse(value="same"), {"value": "same"}),
+        (_ComparableResponse(value="same"), _DifferentResponse(value="same")),
+        (True, 1),
+        (1, 1.0),
+        (["same"], ("same",)),
+        ({"value": "same"}, MappingProxyType({"value": "same"})),
+        ({True: "same"}, {1: "same"}),
+    ),
+    ids=("model-dict", "model-class", "bool-int", "int-float", "list-tuple", "mapping-class", "key-type"),
+)
+def test_model_parity_rejects_nested_type_changes(baseline: object, candidate: object) -> None:
+    with pytest.raises(AssertionError, match=r"\$\.value\[0\]"):
+        assert_model_parity(_PublicValue(value=[baseline]), _PublicValue(value=[candidate]))
+
+
+def test_model_parity_ignores_nested_private_attributes() -> None:
+    baseline: Final = _ComparableResponse(value="same")
+    candidate: Final = _ComparableResponse(value="same")
+    baseline.set_hidden_param("request_id", "baseline")
+    candidate.set_hidden_param("request_id", "candidate")
+
+    assert_model_parity(_PublicValue(value={"nested": [baseline]}), _PublicValue(value={"nested": [candidate]}))
+
+
+@pytest.mark.parametrize("extras", ({"provider_value": "changed"}, {}, {"provider_value": {"value": "same"}}))
+def test_model_parity_compares_public_extras(extras: dict[str, object]) -> None:
+    baseline: Final = _PublicValue.model_validate({"value": None, "provider_value": _ComparableResponse(value="same")})
+    candidate: Final = _PublicValue.model_validate({"value": None, **extras})
+
+    with pytest.raises(AssertionError):
+        assert_model_parity(baseline, candidate)
+
+
+def test_serialized_parity_rejects_boolean_integer_substitution() -> None:
+    with pytest.raises(AssertionError, match="type mismatch"):
+        assert_parity(
+            _execution(body={"enabled": True}, user_agent=SENTINEL),
+            _execution(body={"enabled": 1}, user_agent="candidate"),
+            SENTINEL,
+        )

tests/route_parity/test_stream.py

@@ -2,14 +2,25 @@ from __future__ import annotations
 
 import queue
 from collections.abc import AsyncIterator, Iterator
-from typing import Final
+from typing import Final, Literal, NoReturn
 
 import pytest
-from pydantic import BaseModel, PrivateAttr
+from pydantic import BaseModel, PrivateAttr, ValidationError
 
-from tests.route_parity.models import SDKBytesChunk, SDKJsonChunk, SDKStreamFailed, SDKStreamReport
+from tests.route_parity.models import (
+    SDKBytesChunk,
+    SDKError,
+    SDKJsonChunk,
+    SDKReport,
+    SDKStreamCompleted,
+    SDKStreamFailed,
+    SDKStreamReport,
+    sdk_error_report,
+)
 from tests.route_parity.stream import (
+    StreamCompleted,
     StreamFailed,
+    StreamOutcome,
     assert_stream_parity,
     capture_async_stream,
     capture_sync_stream,
@@ -28,6 +39,10 @@ class _Chunk(BaseModel):
         self._hidden_params[key] = value
 
 
+class _NestedChunk(BaseModel):
+    value: object
+
+
 class _SyncStream:
     def __init__(self, chunks: tuple[object, ...], error: BaseException | None = None) -> None:
         self.chunks: Final = chunks
@@ -53,15 +68,26 @@ class _AsyncStream:
 
 class _PublicStreamError(Exception):
     def __init__(
-        self, message: str = "runtime-specific message", *, status_code: int, llm_provider: str, model: str
+        self,
+        message: str = "invalid input",
+        *,
+        status_code: int = 400,
+        llm_provider: str = "test",
+        model: str = "test-model",
+        code: str = "invalid_input",
+        error_type: str = "validation_error",
+        param: str = "input",
     ) -> None:
         super().__init__(message)
         self.status_code: Final = status_code
         self.llm_provider: Final = llm_provider
         self.model: Final = model
+        self.code: Final = code
+        self.type: Final = error_type
+        self.param: Final = param
 
 
-def _creation_error() -> _SyncStream:
+def _creation_error() -> NoReturn:
     raise _PublicStreamError(status_code=429, llm_provider="test", model="test-model")
 
 
@@ -69,6 +95,22 @@ async def _async_stream(chunks: tuple[object, ...], error: BaseException | None
     return _AsyncStream(chunks, error)
 
 
+async def _consume(
+    mode: Literal["sync", "async"], chunks: tuple[object, ...], error: Exception | None = None
+) -> StreamOutcome:
+    if mode == "sync":
+        return consume_sync_stream(lambda: _SyncStream(chunks, error))
+    return await consume_async_stream(lambda: _async_stream(chunks, error))
+
+
+async def _capture(
+    mode: Literal["sync", "async"], chunks: tuple[object, ...], error: Exception | None = None
+) -> SDKReport:
+    if mode == "sync":
+        return capture_sync_stream(lambda: _SyncStream(chunks, error))
+    return await capture_async_stream(lambda: _async_stream(chunks, error))
+
+
 def test_sync_stream_parity_compares_chunks_and_ignores_private_attrs() -> None:
     python_chunk: Final = _Chunk(value="same")
     accelerated_chunk: Final = _Chunk(value="same")
@@ -120,10 +162,10 @@ def test_stream_outcome_distinguishes_creation_and_iteration_errors() -> None:
 @pytest.mark.asyncio
 async def test_async_stream_uses_same_trace_contract() -> None:
     python_error: Final = _PublicStreamError(
-        "python runtime detail", status_code=500, llm_provider="test", model="test-model"
+        "invalid input\nTraceback (most recent call last):\npython detail", status_code=500
     )
     accelerated_error: Final = _PublicStreamError(
-        "rust runtime detail", status_code=500, llm_provider="test", model="test-model"
+        "invalid input\nTraceback (most recent call last):\nrust detail", status_code=500
     )
     python: Final = await consume_async_stream(lambda: _async_stream((_Chunk(value="same"),), python_error))
     accelerated: Final = await consume_async_stream(lambda: _async_stream((_Chunk(value="same"),), accelerated_error))
@@ -191,3 +233,112 @@ async def test_capture_async_stream_serializes_message_bytes_in_order() -> None:
         b"first",
         b"second",
     )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ("sync", "async"))
+@pytest.mark.parametrize(
+    "candidate_error",
+    (
+        ValueError("invalid input"),
+        _PublicStreamError("changed message"),
+        _PublicStreamError(status_code=429),
+        _PublicStreamError(code="changed_code"),
+        _PublicStreamError(error_type="changed_type"),
+        _PublicStreamError(param="changed_param"),
+        _PublicStreamError(model="changed_model"),
+        _PublicStreamError(llm_provider="changed_provider"),
+    ),
+    ids=("exception", "message", "status", "code", "type", "param", "model", "provider"),
+)
+async def test_stream_parity_rejects_public_error_changes(
+    mode: Literal["sync", "async"], candidate_error: Exception
+) -> None:
+    chunks: Final = (_Chunk(value="partial"),)
+    baseline: Final = await _consume(mode, chunks, _PublicStreamError())
+    candidate: Final = await _consume(mode, chunks, candidate_error)
+
+    with pytest.raises(AssertionError):
+        assert_stream_parity(baseline, candidate)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ("sync", "async"))
+@pytest.mark.parametrize("candidate", (("one",), ("one", "two", "three"), ("two", "one"), ("one", "changed")))
+async def test_stream_parity_checks_event_sequence(mode: Literal["sync", "async"], candidate: tuple[str, ...]) -> None:
+    baseline: Final = await _consume(mode, (_Chunk(value="one"), _Chunk(value="two")))
+    changed: Final = await _consume(mode, tuple(_Chunk(value=value) for value in candidate))
+
+    with pytest.raises(AssertionError):
+        assert_stream_parity(baseline, changed)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ("sync", "async"))
+async def test_stream_parity_preserves_nested_types_and_ignores_private_fields(mode: Literal["sync", "async"]) -> None:
+    first: Final = _Chunk(value="same")
+    second: Final = _Chunk(value="same")
+    first.set_hidden_param("request_id", "first")
+    second.set_hidden_param("request_id", "second")
+    baseline: Final = await _consume(mode, (_NestedChunk(value=[first]),))
+    candidate: Final = await _consume(mode, (_NestedChunk(value=[second]),))
+
+    assert_stream_parity(baseline, candidate)
+    changed: Final = await _consume(mode, (_NestedChunk(value=[{"value": "same"}]),))
+    with pytest.raises(AssertionError, match=r"\$\.chunks\[0\]\.value\[0\]"):
+        assert_stream_parity(baseline, changed)
+
+
+def test_stream_parity_rejects_wrapper_and_chunk_type_changes() -> None:
+    baseline: Final = consume_sync_stream(lambda: _SyncStream((_Chunk(value="same"),)))
+    different_wrapper: Final = consume_sync_stream(lambda: iter((_Chunk(value="same"),)))
+    different_chunk: Final = consume_sync_stream(lambda: _SyncStream(({"value": "same"},)))
+
+    with pytest.raises(AssertionError):
+        assert_stream_parity(baseline, different_wrapper)
+    with pytest.raises(AssertionError):
+        assert_stream_parity(baseline, different_chunk)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ("sync", "async"))
+@pytest.mark.parametrize("error", (None, _PublicStreamError()))
+async def test_capture_keeps_serialization_failures_out_of_sdk_errors(
+    mode: Literal["sync", "async"], error: Exception | None
+) -> None:
+    with pytest.raises(ValidationError):
+        await _capture(mode, (_Chunk(value="valid"), object()), error)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ("sync", "async"))
+async def test_empty_stream_completes(mode: Literal["sync", "async"]) -> None:
+    outcome: Final = await _consume(mode, ())
+    assert outcome.chunks == ()
+    assert outcome.terminal == StreamCompleted()
+    assert await _capture(mode, ()) == SDKStreamReport(chunks=(), terminal=SDKStreamCompleted())
+
+
+@pytest.mark.asyncio
+async def test_async_creation_error_matches_sync_capture() -> None:
+    async def create() -> _AsyncStream:
+        _creation_error()
+
+    sync: Final = consume_sync_stream(_creation_error)
+    asynchronous: Final = await consume_async_stream(create)
+    assert_stream_parity(sync, asynchronous)
+    assert isinstance(sync.terminal, StreamFailed)
+    assert sync.terminal.phase == "creation"
+    assert isinstance(capture_sync_stream(_creation_error), SDKError)
+    assert capture_sync_stream(_creation_error) == await capture_async_stream(create) == sync.terminal.error
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("mode", ("sync", "async"))
+async def test_capture_preserves_partial_output_and_complete_error(mode: Literal["sync", "async"]) -> None:
+    error: Final = _PublicStreamError()
+    report: Final = await _capture(mode, (_Chunk(value="partial"),), error)
+    assert report == SDKStreamReport(
+        chunks=(SDKJsonChunk(value={"value": "partial"}),),
+        terminal=SDKStreamFailed(error=sdk_error_report(error)),
+    )