Files
turnstone/tests/test_perception.py
T
Patrick Buckley 1e7ad7bcb6 feat(providers): one transport — drain create_streaming, retire create_completion (#831)
Every single-shot lane (model_turn: judges, titles, compaction, web-fetch
extraction, perception, eval, optimizer) now samples through the provider's
streaming entry and accumulates via a shared drain_stream(), deleting
create_completion from the Protocol and all three adapters (xai/google
inherit). Request shaping can no longer drift between the two consumption
styles, and callers keep the exact CompletionResult contract.

The drain mirrors the main loop's proven chunk semantics: per-field
max-merge for usage (Anthropic splits prompt/completion across
message_start/message_delta), tool-call assembly by delta index,
provider_blocks from the terminal emission, trailing citation info folded
back into content (byte-matching the old format_citations append),
mid-stream status pings dropped.

Also in this change:

- model_turn grows cancel_ref; both judges wire their run_with_deadline
  abandon paths to a new StreamAbortRef (deadline.py) that closes the SDK
  stream — a timed-out judge call now aborts its HTTP read instead of
  pinning a daemon thread until the next upstream chunk. The append hook
  covers the arrival race, mirroring ChatSession._CancelRef.
- Responses streaming gains the response.incomplete terminal handler
  (truncated runs were mislabeled finish=stop and lost final usage AND
  collected provider_blocks) and a refusal handler ([Refused: …] content,
  matching the retired non-streaming rendering). Both also fix the main
  chat loop, which shared the gaps.
- supports_streaming capability flag deleted (zero readers) along with
  its admin capability tile; o1-era models that reject streaming need a
  model alias pointing at a current model (release-noted).
- Helpers that existed only for the deleted transport go with it:
  Responses._parse_response, chat/google._extract_tool_calls.

Known behavioral deltas (release-noted): OpenAI-compatible servers that
ignore stream_options.include_usage stop producing usage rows on these
lanes; multiple Anthropic text blocks concatenate without the old "\n"
joint (matching the main loop); model_turn lanes no longer risk client
read-timeouts on long generations — the reason the Anthropic adapter
already drained a stream internally.

Tests: new test_drain_stream.py pins the accumulator rules; shared fakes
(as_stream, fake_chat_stream, fake_anthropic_stream) migrate 11 suites to
the streaming transport, with the task-agent and adapter suites now
exercising the real _iter_stream + drain path end to end.
2026-07-13 22:39:19 -07:00

142 lines
4.9 KiB
Python

"""Unit tests for the perception wire-fallback (turnstone/core/perception.py)."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import pytest
from tests._session_helpers import as_stream, mock_completion_result
from turnstone.core import perception
if TYPE_CHECKING:
from collections.abc import Iterator
class _StubProvider:
"""Minimal LLMProvider stand-in: counts calls, can fail the first N.
``describe`` routes through ``model_turn``, so the stub carries the lane
surface (``provider_name``, ``get_capabilities``) and returns a full
``CompletionResult`` shape, and it records the ``resolve_attachments``
callback the translator would use to materialize the by-reference parts.
"""
provider_name = "openai-compatible"
def __init__(self, *, content: str = "a description", fail_times: int = 0) -> None:
self.calls = 0
self._content = content
self._fail_times = fail_times
self.last_messages: list[dict[str, Any]] | None = None
self.last_resolve: Any = None
def get_capabilities(self, model: str) -> Any:
from turnstone.core.providers._protocol import ModelCapabilities
return ModelCapabilities()
def create_streaming(
self,
*,
client: Any,
model: str,
messages: list[dict[str, Any]],
resolve_attachments: Any = None,
**_: Any,
) -> Any:
self.calls += 1
self.last_messages = messages
self.last_resolve = resolve_attachments
if self.calls <= self._fail_times:
raise RuntimeError("backend down")
# Shared field inventory: when model_turn's re-ingest reads a new
# CompletionResult field, mock_completion_result is the ONE
# definition to extend and this suite moves with it.
return as_stream(mock_completion_result(self._content))
@pytest.fixture(autouse=True)
def _clear_cache() -> Iterator[None]:
perception._clear_perception_cache_for_test()
yield
perception._clear_perception_cache_for_test()
def _parts() -> list[dict[str, Any]]:
return [{"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}]
def test_describe_lowers_prompt_then_by_reference_parts() -> None:
prov = _StubProvider(content="desc")
out = perception.describe(provider=prov, client=object(), model="m", parts=_parts()) # type: ignore[arg-type]
assert out == "desc"
assert prov.last_messages is not None
content = prov.last_messages[0]["content"]
assert content[0]["type"] == "text" # prompt leads
# The attachment rides by reference; the translator materializes it via
# the threaded resolver, which must return the prebuilt parts verbatim.
assert content[1]["attachment_id"] == "perception-input"
assert prov.last_resolve is not None
assert prov.last_resolve(["perception-input"]) == {"perception-input": _parts()}
def test_describe_empty_parts_skips_backend() -> None:
prov = _StubProvider()
assert perception.describe(provider=prov, client=object(), model="m", parts=[]) == "" # type: ignore[arg-type]
assert prov.calls == 0
def test_describe_cached_memoizes_by_alias_and_hash() -> None:
prov = _StubProvider(content="desc")
kw: dict[str, Any] = {
"provider": prov,
"client": object(),
"model": "m",
"alias": "omni",
"content_hash": "h1",
"parts": _parts(),
}
assert perception.describe_cached(**kw) == "desc"
assert perception.describe_cached(**kw) == "desc"
assert prov.calls == 1 # second served from cache
perception.describe_cached(**{**kw, "content_hash": "h2"})
assert prov.calls == 2 # distinct hash → fresh perceive
def test_describe_cached_does_not_cache_failures() -> None:
prov = _StubProvider(content="recovered", fail_times=1)
kw: dict[str, Any] = {
"provider": prov,
"client": object(),
"model": "m",
"alias": "omni",
"content_hash": "h",
"parts": _parts(),
}
assert perception.describe_cached(**kw) == "" # backend down → "" (uncached)
assert perception.describe_cached(**kw) == "recovered" # retried, succeeds
assert prov.calls == 2
def test_describe_peek_returns_none_when_absent() -> None:
assert perception.describe_peek(alias="omni", content_hash="missing") is None
def test_describe_peek_returns_cached_without_recompute() -> None:
prov = _StubProvider(content="desc")
kw: dict[str, Any] = {
"provider": prov,
"client": object(),
"model": "m",
"alias": "omni",
"content_hash": "h",
"parts": _parts(),
}
perception.describe_cached(**kw) # populate the memo
assert prov.calls == 1
# Peek serves the memoized text and never re-invokes the backend — this is
# what lets the wire resolver skip the PDF rasterize on a cross-send hit.
assert perception.describe_peek(alias="omni", content_hash="h") == "desc"
assert prov.calls == 1