mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
7da07e2350
_resolve_attachments re-runs on every agentic round-trip (and per fallback model), each time re-fetching every attachment across the full history and re-rasterizing / re-base64'ing it. A 10-page PDF in a 10-cycle tool turn was rendered dozens of times. Add a per-send memo (self._wire_part_cache) keyed by (attachment_id, caps-signature): the materialized wire part is computed at most once per send. The cache is None outside a send (display/export paths unaffected) and reset per send to bound the heavy rasterized-page parts and pick up any mid-session capability change. Skip the DB fetch entirely when every id is already cached. Also peek the perception (alias, content_hash) memo before building parts in _perception_fallback_part, so a cross-send describe hit no longer wastes a PDF rasterize. Leaves pdf.py's deliberate no-module-cache stance intact — the per-send scope addresses the round-trip amplification without the durable store it defers. Adds describe_peek() + per-send-cache and peek tests.
114 lines
3.8 KiB
Python
114 lines
3.8 KiB
Python
"""Unit tests for the perception wire-fallback (turnstone/core/perception.py)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
import pytest
|
|
|
|
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."""
|
|
|
|
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
|
|
|
|
def create_completion(
|
|
self, *, client: Any, model: str, messages: list[dict[str, Any]], **_: Any
|
|
) -> SimpleNamespace:
|
|
self.calls += 1
|
|
self.last_messages = messages
|
|
if self.calls <= self._fail_times:
|
|
raise RuntimeError("backend down")
|
|
return SimpleNamespace(content=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_builds_prompt_then_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
|
|
assert content[1]["type"] == "image_url" # attachment parts follow
|
|
|
|
|
|
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
|