mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
98e96ab5f3
* feat(models): add per-alias concurrency admission Add registry-backed FIFO admission limits with queue-aware deadlines and full-stream leases. Expose max_concurrency through storage, admin configuration, OpenAPI, documentation, and diagrams, with role and live backend count coverage. * fix(api): omit null concurrency schema default Keep max_concurrency optional for presence-keyed updates without advertising a null default for its non-null integer OpenAPI shape.
376 lines
12 KiB
Python
376 lines
12 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
|
|
from turnstone.core.model_turn import ModelLane, ResolvedModelBinding, resolve_lane
|
|
|
|
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 that ``model_turn`` already
|
|
materialized by-reference parts and cleared the provider callback.
|
|
"""
|
|
|
|
provider_name = "openai-compatible"
|
|
retryable_error_names: frozenset[str] = frozenset()
|
|
|
|
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 convert_tools(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
return tools
|
|
|
|
def extract_reasoning_text(self, provider_blocks: list[dict[str, Any]] | None) -> str:
|
|
return ""
|
|
|
|
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 _lane(provider: _StubProvider, *, alias: str = "omni") -> ModelLane:
|
|
"""Build the same resolved binding snapshot production hands perception."""
|
|
return resolve_lane(provider, object(), "m", alias=alias)
|
|
|
|
|
|
def _binding(
|
|
provider: _StubProvider,
|
|
*,
|
|
alias: str = "omni",
|
|
generation: int = 0,
|
|
) -> ResolvedModelBinding:
|
|
return ResolvedModelBinding(
|
|
lane=_lane(provider, alias=alias),
|
|
config=None,
|
|
registry_generation=generation,
|
|
)
|
|
|
|
|
|
def test_describe_lowers_prompt_then_by_reference_parts() -> None:
|
|
prov = _StubProvider(content="desc")
|
|
out = perception.describe(lane=_lane(prov), parts=_parts())
|
|
assert out == "desc"
|
|
assert prov.last_messages is not None
|
|
content = prov.last_messages[0]["content"]
|
|
assert content[0]["type"] == "text" # prompt leads
|
|
# model_turn materializes before admission, then hands the provider the
|
|
# prebuilt inline part with no resolver left to invoke under the gate.
|
|
assert content[1] == _parts()[0]
|
|
assert prov.last_resolve is None
|
|
|
|
|
|
def test_describe_empty_parts_skips_backend() -> None:
|
|
prov = _StubProvider()
|
|
assert perception.describe(lane=_lane(prov), parts=[]) == ""
|
|
assert prov.calls == 0
|
|
|
|
|
|
def test_describe_passes_the_exact_supplied_lane_to_model_turn(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from turnstone.core.model_turn import ModelTurnResult
|
|
from turnstone.core.trajectory import Turn
|
|
|
|
binding = _binding(_StubProvider())
|
|
lane = binding.lane
|
|
seen: list[ModelLane] = []
|
|
|
|
def _sample(sample_lane: ModelLane, *_args: Any, **_kwargs: Any) -> ModelTurnResult:
|
|
seen.append(sample_lane)
|
|
return ModelTurnResult(
|
|
turn=Turn.assistant("from seam"),
|
|
finish_reason="stop",
|
|
usage=None,
|
|
tool_calls=[],
|
|
)
|
|
|
|
monkeypatch.setattr(perception, "model_turn", _sample)
|
|
|
|
assert perception.describe(lane=lane, parts=_parts()) == "from seam"
|
|
assert seen == [lane]
|
|
assert seen[0] is lane
|
|
|
|
|
|
def test_cancellation_ref_reaches_model_turn_and_is_not_swallowed(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from turnstone.core.deadline import DeadlineCancelledError
|
|
|
|
ref = object()
|
|
seen: list[Any] = []
|
|
|
|
def abort(*_args: Any, **kwargs: Any) -> Any:
|
|
seen.append(kwargs.get("cancel_ref"))
|
|
raise DeadlineCancelledError("stopped")
|
|
|
|
monkeypatch.setattr(perception, "model_turn", abort)
|
|
binding = _binding(_StubProvider())
|
|
lane = binding.lane
|
|
|
|
with pytest.raises(DeadlineCancelledError, match="stopped"):
|
|
perception.describe(lane=lane, parts=_parts(), cancel_ref=ref)
|
|
with pytest.raises(DeadlineCancelledError, match="stopped"):
|
|
perception.describe_cached(
|
|
binding=binding,
|
|
principal_id="user-a",
|
|
content_hash="h-cancel",
|
|
parts=_parts(),
|
|
cancel_ref=ref,
|
|
)
|
|
|
|
assert seen == [ref, ref]
|
|
assert (
|
|
perception.describe_peek(
|
|
principal_id="user-a",
|
|
binding=binding,
|
|
content_hash="h-cancel",
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_completed_cancelled_description_is_not_memoized(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
from turnstone.core.deadline import DeadlineCancelledError, StreamAbortRef
|
|
|
|
binding = _binding(_StubProvider())
|
|
cancelled_ref = StreamAbortRef()
|
|
calls: list[str] = []
|
|
|
|
def complete_after_cancel(**_kwargs: Any) -> str:
|
|
calls.append("cancelled")
|
|
cancelled_ref.abort()
|
|
return "late description"
|
|
|
|
monkeypatch.setattr(perception, "describe", complete_after_cancel)
|
|
with pytest.raises(DeadlineCancelledError):
|
|
perception.describe_cached(
|
|
binding=binding,
|
|
principal_id="user-a",
|
|
content_hash="late",
|
|
parts=_parts(),
|
|
cancel_ref=cancelled_ref,
|
|
)
|
|
assert (
|
|
perception.describe_peek(
|
|
principal_id="user-a",
|
|
binding=binding,
|
|
content_hash="late",
|
|
)
|
|
is None
|
|
)
|
|
|
|
monkeypatch.setattr(
|
|
perception,
|
|
"describe",
|
|
lambda **_kwargs: calls.append("fresh") or "fresh description",
|
|
)
|
|
assert (
|
|
perception.describe_cached(
|
|
binding=binding,
|
|
principal_id="user-a",
|
|
content_hash="late",
|
|
parts=_parts(),
|
|
cancel_ref=StreamAbortRef(),
|
|
)
|
|
== "fresh description"
|
|
)
|
|
assert calls == ["cancelled", "fresh"]
|
|
|
|
|
|
def test_describe_cached_memoizes_by_principal_alias_generation_and_hash() -> None:
|
|
prov = _StubProvider(content="desc")
|
|
binding = _binding(prov)
|
|
kw: dict[str, Any] = {
|
|
"binding": binding,
|
|
"principal_id": "user-a",
|
|
"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
|
|
perception.describe_cached(**{**kw, "principal_id": "user-b"})
|
|
assert prov.calls == 3 # same content under another user's grant → fresh perceive
|
|
perception.describe_cached(**{**kw, "binding": _binding(prov, alias="other")})
|
|
assert prov.calls == 4 # same content under another alias → fresh perceive
|
|
newer = _binding(prov, generation=1)
|
|
perception.describe_cached(**{**kw, "binding": newer})
|
|
assert prov.calls == 5 # same alias under a new registry generation → fresh perceive
|
|
assert (
|
|
perception.describe_peek(
|
|
principal_id="user-a",
|
|
binding=binding,
|
|
content_hash="h1",
|
|
)
|
|
== "desc"
|
|
)
|
|
assert (
|
|
perception.describe_peek(
|
|
principal_id="user-a",
|
|
binding=_binding(prov, generation=2),
|
|
content_hash="h1",
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_describe_cached_does_not_cache_failures() -> None:
|
|
prov = _StubProvider(content="recovered", fail_times=1)
|
|
kw: dict[str, Any] = {
|
|
"binding": _binding(prov),
|
|
"principal_id": "user-a",
|
|
"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:
|
|
binding = _binding(_StubProvider())
|
|
assert (
|
|
perception.describe_peek(
|
|
principal_id="user-a",
|
|
binding=binding,
|
|
content_hash="missing",
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_describe_peek_returns_cached_without_recompute() -> None:
|
|
prov = _StubProvider(content="desc")
|
|
binding = _binding(prov)
|
|
kw: dict[str, Any] = {
|
|
"binding": binding,
|
|
"principal_id": "user-a",
|
|
"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(
|
|
principal_id="user-a",
|
|
binding=binding,
|
|
content_hash="h",
|
|
)
|
|
== "desc"
|
|
)
|
|
assert (
|
|
perception.describe_peek(
|
|
principal_id="user-b",
|
|
binding=binding,
|
|
content_hash="h",
|
|
)
|
|
is None
|
|
)
|
|
assert prov.calls == 1
|
|
|
|
|
|
def test_describe_cached_memoizes_empty_descriptions() -> None:
|
|
# A completed-but-empty description (an all-reasoning pass) memoizes
|
|
# like any other result: one perceive per key, ever — bounded cost.
|
|
# The pin-until-restart residual is deliberate; the remediation is
|
|
# server-side (reasoning parser / template thinking toggle).
|
|
prov = _StubProvider(content="")
|
|
binding = _binding(prov)
|
|
kw: dict[str, Any] = {
|
|
"binding": binding,
|
|
"principal_id": "user-a",
|
|
"content_hash": "h-empty",
|
|
"parts": _parts(),
|
|
}
|
|
assert perception.describe_cached(**kw) == ""
|
|
assert perception.describe_cached(**kw) == ""
|
|
assert prov.calls == 1 # second call served from the memo
|
|
assert (
|
|
perception.describe_peek(
|
|
principal_id="user-a",
|
|
binding=binding,
|
|
content_hash="h-empty",
|
|
)
|
|
== ""
|
|
)
|
|
|
|
|
|
def test_racing_empty_result_never_clobbers_memoized_real_description(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
# The describe call runs unlocked: a racer can memoize a REAL
|
|
# description while another call is producing "". The empty commit
|
|
# must yield to the existing memo, never overwrite it.
|
|
binding = _binding(_StubProvider(content=""))
|
|
call_key = {"principal_id": "user-a", "content_hash": "h-race"}
|
|
cache_key = {**call_key, "binding": binding}
|
|
|
|
def _racing_describe(**_kw: Any) -> str:
|
|
with perception._cache_lock:
|
|
perception._cache[perception._cache_key(**cache_key)] = "real from racer"
|
|
return ""
|
|
|
|
monkeypatch.setattr(perception, "describe", _racing_describe)
|
|
out = perception.describe_cached(
|
|
binding=binding,
|
|
parts=_parts(),
|
|
**call_key,
|
|
)
|
|
assert out == "real from racer"
|
|
assert (
|
|
perception.describe_peek(**cache_key) == "real from racer"
|
|
) # the billed real description survived
|