mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
ffe8214cfe
Proves the effort-ladder projection against the real request path instead of against the mapping helpers it shares with it. For 22 (provider lane x capability shape) points — both Anthropic lanes, openai-compatible on both API surfaces, openai, google (default and template-override hybrid), xai (default and inert-override), and the DeepSeek/qwen template contracts — every knob position is driven through the actual provider create_streaming against a recording fake client, and two invariants are asserted per shape: 1. each ladder token decodes to an expected effort wire subset (toggle / template effort / flat param / thinking budget / output_config) that must equal the captured kwargs exactly; 2. two knob positions carry equal tokens iff they produce identical effort-relevant wire payloads — the grouping promise the UI annotations lean on. The RecordingClient SDK-seam stub moves from the wire-payload golden harness into tests/_wire_capture.py so both suites capture at the same seam. Verified the harness catches the bug class it was built for: re-adding xai to _CHAT_LANES fails xai-template-override-inert.
77 lines
2.2 KiB
Python
77 lines
2.2 KiB
Python
"""Recording fake SDK client — captures the kwargs at each provider's seam.
|
|
|
|
Every provider's ``create_streaming`` assembles its kwargs and calls the
|
|
SDK *eagerly* before returning the stream iterator (Anthropic
|
|
``client.messages.stream``, OpenAI ``client.chat.completions.create``,
|
|
Responses ``client.responses.create/stream``), so driving a provider
|
|
against a :class:`RecordingClient` captures the full composed request
|
|
payload without a network round-trip.
|
|
|
|
Shared by the wire-payload golden harness (``test_wire_payload_golden``)
|
|
and the effort-ladder parity harness (``test_effort_ladder_wire_parity``)
|
|
so both assert against the same capture seam.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Iterator
|
|
|
|
|
|
class _EmptyStream:
|
|
"""Stand-in for an SDK stream / stream-manager: empty iterable AND no-op CM."""
|
|
|
|
def __iter__(self) -> Iterator[Any]:
|
|
return iter(())
|
|
|
|
def __enter__(self) -> _EmptyStream:
|
|
return self
|
|
|
|
def __exit__(self, *exc: object) -> None:
|
|
return None
|
|
|
|
|
|
class _Seam:
|
|
"""Records the kwargs of a single SDK call, returns an empty stream stub."""
|
|
|
|
def __init__(self, sink: dict[str, Any]) -> None:
|
|
self._sink = sink
|
|
|
|
def __call__(self, **kwargs: Any) -> _EmptyStream:
|
|
# Last write wins; only one seam is exercised per provider call.
|
|
self._sink["payload"] = kwargs
|
|
return _EmptyStream()
|
|
|
|
|
|
class _Completions:
|
|
def __init__(self, sink: dict[str, Any]) -> None:
|
|
self.create = _Seam(sink)
|
|
|
|
|
|
class _Chat:
|
|
def __init__(self, sink: dict[str, Any]) -> None:
|
|
self.completions = _Completions(sink)
|
|
|
|
|
|
class _Messages:
|
|
def __init__(self, sink: dict[str, Any]) -> None:
|
|
self.stream = _Seam(sink)
|
|
|
|
|
|
class _Responses:
|
|
def __init__(self, sink: dict[str, Any]) -> None:
|
|
self.create = _Seam(sink)
|
|
self.stream = _Seam(sink)
|
|
|
|
|
|
class RecordingClient:
|
|
"""Fake SDK client exposing every provider's call seam, recording kwargs."""
|
|
|
|
def __init__(self) -> None:
|
|
self.captured: dict[str, Any] = {}
|
|
self.messages = _Messages(self.captured)
|
|
self.chat = _Chat(self.captured)
|
|
self.responses = _Responses(self.captured)
|