feat(headroom): allow opt-in compression of the latest turn

This commit is contained in:
Joshua Valluru 2026-10-19 15:18:05 -07:00
parent b110c8cfa7
commit c7ddde6ecb
13 changed files with 404 additions and 34 deletions

View file

@ -222,12 +222,14 @@ def _cached_prefix_indices(messages: Sequence[Mapping[str, object]]) -> tuple[in
return tuple(range(last_breakpoint + 1))
def get_protected_indices(messages: Sequence[Mapping[str, object]]) -> tuple[int, ...]:
def get_protected_indices(
messages: Sequence[Mapping[str, object]], *, protect_latest_turn: bool = True
) -> tuple[int, ...]:
"""
Return indices of messages that must never be compressed:
- All system messages
- The last user message
- The last assistant message
- The last user message when protect_latest_turn is enabled
- The last assistant message when protect_latest_turn is enabled
- Every message up to and including the last one carrying an Anthropic cache_control breakpoint
The last user message is what the model is being asked to act on right now,
@ -240,7 +242,8 @@ def get_protected_indices(messages: Sequence[Mapping[str, object]]) -> tuple[int
system_indices: Final = tuple(index for index, msg in enumerate(messages) if msg.get("role", "") == "system")
last_user: Final = tuple(index for index, msg in enumerate(messages) if msg.get("role", "") == "user")[-1:]
assistant_indices: Final = tuple(index for index, msg in enumerate(messages) if msg.get("role", "") == "assistant")
return tuple(dict.fromkeys(system_indices + last_user + assistant_indices[-1:] + _cached_prefix_indices(messages)))
latest_indices: Final = last_user + assistant_indices[-1:] if protect_latest_turn else ()
return tuple(dict.fromkeys(system_indices + latest_indices + _cached_prefix_indices(messages)))
def _combine_scores(

View file

@ -1126,6 +1126,9 @@ class CustomGuardrail(CustomLogger):
"""
return True
def preserve_request_text_metadata(self) -> bool:
return False
def structured_messages_cover_full_request(self) -> bool:
"""Whether returned ``structured_messages`` span the whole request.

View file

@ -266,13 +266,23 @@ def _item_rewrite_field(item: Mapping[str, object]) -> str | None:
return _PATCHABLE_ITEM_FIELDS.get(item_type)
def _rewritten_input_item(item: Mapping[str, object], rewritten: object) -> Mapping[str, object] | None:
def _rewritten_input_item(
item: Mapping[str, object], rewritten: object, *, preserve_text_metadata: bool = False
) -> Mapping[str, object] | None:
field: Final = _item_rewrite_field(item)
if field is None or not isinstance(rewritten, Mapping):
return None
rewritten_content: Final = rewritten.get("content")
if isinstance(item.get(field), str) and isinstance(rewritten_content, str):
return {**item, field: rewritten_content}
if preserve_text_metadata:
from litellm.proxy.guardrails.guardrail_hooks.content_text import content_to_text, replace_text_content
typed_rewritten: Final = _item_fields_adapter.validate_python(rewritten)
replaced: Final = replace_text_content(
item.get(field), content_to_text(typed_rewritten.get("content"), ignore_empty=True)
)
return {**item, field: replaced} if replaced is not None else None
rewritten_row: Final = cast("AllMessageValues", rewritten) # cast-ok: guardrails hand back chat-shaped rows
converted_items, _ = LiteLLMResponsesTransformationHandler().convert_chat_completion_messages_to_responses_api(
[rewritten_row]
@ -493,12 +503,22 @@ def _input_item_role(item: object) -> str:
return role.lower() if isinstance(role, str) else ""
def _patched_request_fields(
def patch_response_request_fields(
raw_input: object,
instructions: object,
original_messages: Sequence[object],
structured_messages: Sequence[object],
*,
preserve_text_metadata: bool = False,
) -> _RequestFields | None:
if isinstance(raw_input, str) and preserve_text_metadata:
return patch_response_request_fields(
[{"role": "user", "content": raw_input}],
instructions,
original_messages,
structured_messages,
preserve_text_metadata=True,
)
if not isinstance(raw_input, list) or len(original_messages) != len(structured_messages):
return None
offset: Final = 1 if instructions else 0
@ -528,6 +548,7 @@ def _patched_request_fields(
item_for_message[message_index]: _rewritten_input_item(
cast("Mapping[str, object]", raw_input[item_for_message[message_index]]), # cast-ok: checked Mappings
rewritten,
preserve_text_metadata=preserve_text_metadata,
)
for message_index, rewritten in body_changes
}
@ -545,11 +566,15 @@ def _patch_or_convert_request_fields(
instructions: object,
original_messages: Sequence[object],
structured_messages: Sequence[AllMessageValues],
*,
preserve_text_metadata: bool = False,
) -> _RequestFields | None:
if not isinstance(structured_messages, list):
return None
patched: Final = _patched_request_fields(raw_input, instructions, original_messages, structured_messages)
if patched is not None:
patched: Final = patch_response_request_fields(
raw_input, instructions, original_messages, structured_messages, preserve_text_metadata=preserve_text_metadata
)
if patched is not None or preserve_text_metadata:
return patched
input_items, converted_instructions = (
LiteLLMResponsesTransformationHandler().convert_chat_completion_messages_to_responses_api(structured_messages)
@ -762,7 +787,11 @@ class OpenAIResponsesHandler(BaseTranslation):
)
)
return _patch_or_convert_request_fields(
data.get("input"), data.get("instructions"), structured_messages, merged
data.get("input"),
data.get("instructions"),
structured_messages,
merged,
preserve_text_metadata=guardrail_to_apply.preserve_request_text_metadata(),
)
def extract_request_tool_names(self, data: Mapping[str, object]) -> list[str]:

View file

@ -16335,6 +16335,12 @@
"description": "Client secret of the gateway's Entra app registration, used to perform the On-Behalf-Of exchange. Falls back to the AGENT365_CLIENT_SECRET environment variable.",
"title": "Client Secret"
},
"compress_latest_turn": {
"default": false,
"description": "Allow compression of latest user and tool-result text, which can lose instructions or details. System prompts, assistant messages, cached prefixes and retrieved content remain protected. Unsafe rewrites retain the original input.",
"title": "Compress Latest Turn",
"type": "boolean"
},
"confidence_threshold": {
"default": 0.5,
"default_value": 0.5,

View file

@ -12,13 +12,15 @@ therefore only be written back over a run of text parts, never across a
non-text part, which is what ``is_all_text_parts`` gates.
"""
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from typing import Final
from pydantic import TypeAdapter, ValidationError
from litellm.litellm_core_utils.prompt_templates.factory import get_attribute_or_key
def content_to_text(content: object) -> str:
def content_to_text(content: object, *, ignore_empty: bool = False) -> str:
"""Collapse a message ``content`` (str or list-of-parts) to plain text.
For the multimodal list shape, joins ``{type: "text", text: ...}`` parts
@ -31,7 +33,7 @@ def content_to_text(content: object) -> str:
for part in content:
if isinstance(part, dict) and part.get("type") == "text":
text = part.get("text")
if isinstance(text, str):
if isinstance(text, str) and (text or not ignore_empty):
parts.append(text)
return "\n\n".join(parts)
return ""
@ -94,3 +96,23 @@ def assistant_text_from_response(response: object) -> str | None:
if output_parts:
return "".join(output_parts)
return None
_TEXT_PARTS_ADAPTER: Final = TypeAdapter(tuple[Mapping[str, object], ...])
def replace_text_content(content: object, text: str) -> object:
if isinstance(content, str):
return text
if not isinstance(content, list) or not content:
return None
try:
parts: Final = _TEXT_PARTS_ADAPTER.validate_python(content)
except ValidationError:
return None
if any(
part.get("type") not in ("text", "input_text", "output_text") or not isinstance(part.get("text"), str)
for part in parts
):
return None
return [{**part, "text": text if index == 0 else ""} for index, part in enumerate(parts)]

View file

@ -37,6 +37,7 @@ def initialize_guardrail(litellm_params: LitellmParams, guardrail: Guardrail) ->
unreachable_fallback=litellm_params.unreachable_fallback,
timeout=litellm_params.timeout,
ccr_retrieval=litellm_params.ccr_retrieval,
compress_latest_turn=litellm_params.compress_latest_turn,
)
litellm.logging_callback_manager.add_litellm_callback( # pyright: ignore[reportUnknownMemberType]
_callback

View file

@ -37,6 +37,7 @@ from litellm.proxy.guardrails.guardrail_hooks.content_text import (
content_to_text,
is_all_text_parts,
merge_rewritten_text_parts,
replace_text_content,
)
from litellm.proxy.spend_tracking.compression_savings import HEADROOM_GUARDRAIL_PROVIDER
from litellm.secret_managers.main import get_secret_str
@ -141,6 +142,31 @@ def _restore_content_shapes(
return restored
def _compressed_message_content(original: Mapping[str, object], rewritten: Mapping[str, object]) -> object:
content: Final = rewritten.get("content")
if content == original.get("content"):
return content
if not isinstance(content, str):
return None
return replace_text_content(original.get("content"), content)
def _restore_compressed_text(
originals: Sequence[Mapping[str, object]], returned: Sequence[Mapping[str, object]]
) -> tuple[Mapping[str, object], ...] | None:
if any(
original.get("role") != rewritten.get("role") or original.get("tool_call_id") != rewritten.get("tool_call_id")
for original, rewritten in zip(originals, returned)
):
return None
contents: Final = tuple(
_compressed_message_content(original, rewritten) for original, rewritten in zip(originals, returned)
)
if any(content is None for content in contents):
return None
return tuple({**original, "content": content} for original, content in zip(originals, contents))
def _tool_call_name(tool_call: Mapping[str, object]) -> str | None:
function: Final = tool_call.get("function")
if not _is_str_object_dict(function):
@ -239,7 +265,10 @@ def _retrieval_result_indices(
def _protected_indices(
messages: Sequence[Mapping[str, object]], extra_retrieve_call_ids: frozenset[str] = frozenset()
messages: Sequence[Mapping[str, object]],
extra_retrieve_call_ids: frozenset[str] = frozenset(),
*,
compress_latest_turn: bool = False,
) -> frozenset[int]:
"""Indices headroom must not send to the compression service.
@ -257,9 +286,9 @@ def _protected_indices(
so the model's own earlier tables came back rewritten and it imitated the
shape. The tool results those turns asked for stay compressible.
"""
protected: Final = frozenset(get_protected_indices(messages)) | _retrieval_result_indices(
messages, extra_retrieve_call_ids
)
protected: Final = frozenset(
get_protected_indices(messages, protect_latest_turn=not compress_latest_turn)
) | _retrieval_result_indices(messages, extra_retrieve_call_ids)
return (
protected
| frozenset(
@ -273,10 +302,10 @@ def _protected_indices(
def _restore_protected_messages(
messages: Sequence[dict[str, object]],
compressed: Sequence[dict[str, object]],
messages: Sequence[Mapping[str, object]],
compressed: Sequence[Mapping[str, object]],
protected_indices: frozenset[int],
) -> Sequence[dict[str, object]]:
) -> Sequence[Mapping[str, object]]:
"""Put the rows that were held back from compression at their original positions.
Requires one returned row per row actually sent, which ``_call_compress``
@ -505,6 +534,7 @@ class HeadroomGuardrail(CustomGuardrail):
unreachable_fallback: str | None = None,
timeout: float | None = None,
ccr_retrieval: bool = True,
compress_latest_turn: bool = False,
):
self.headroom_api_base = (api_base or get_secret_str("HEADROOM_API_BASE") or "").rstrip("/")
if not self.headroom_api_base:
@ -518,6 +548,7 @@ class HeadroomGuardrail(CustomGuardrail):
"fail_open" if unreachable_fallback == "fail_open" else "fail_closed"
)
self.ccr_retrieval = ccr_retrieval
self.compress_latest_turn = compress_latest_turn
self.async_handler = get_async_httpx_client(
llm_provider=httpxSpecialProvider.GuardrailCallback,
)
@ -530,6 +561,9 @@ class HeadroomGuardrail(CustomGuardrail):
)
self.timeout = self._resolve_timeout(timeout)
def preserve_request_text_metadata(self) -> bool:
return self.compress_latest_turn
def _should_bypass(self, request_data: dict) -> bool:
psr: Final = request_data.get("proxy_server_request")
if not _is_str_object_dict(psr):
@ -677,6 +711,9 @@ class HeadroomGuardrail(CustomGuardrail):
)
filtered: Final = [item for item in compressed_messages if _is_str_object_dict(item)]
if self.compress_latest_turn and (len(filtered) != len(messages) or len(filtered) != len(compressed_messages)):
return _CompressResult(messages, False, {})
if not filtered:
return _CompressResult(
self._handle_compress_failure(
@ -800,20 +837,12 @@ class HeadroomGuardrail(CustomGuardrail):
if not messages:
return inputs
# The last user message is the instruction the model is being asked to
# act on, so replacing it with a marker means the model answers a
# retrieval result instead of the request. Protected rows are held back
# from the payload rather than pinned after the fact, so their tokens
# are not counted as savings we never apply; the Anthropic write-back
# discards a compressed system prompt outright. Keep it that way unless
# /v1/compress grows a field for sending the live turn as the retrieval
# query without compressing it: query-aware compression reads the newest
# user message, so it is withheld here at some cost to history ranking.
# request_data is a bare dict on the base signature; narrow it before
# reading the untranslated messages so long tool names can be recovered.
raw_messages: Final = _REQUEST_DATA_ADAPTER.validate_python(request_data).get("messages")
typed_request: Final = _REQUEST_DATA_ADAPTER.validate_python(request_data)
raw_messages: Final = typed_request.get("messages")
raw_retrieve_call_ids: Final = _raw_retrieve_call_ids(raw_messages)
protected_indices: Final = _protected_indices(messages, raw_retrieve_call_ids)
protected_indices: Final = _protected_indices(
messages, raw_retrieve_call_ids, compress_latest_turn=self.compress_latest_turn
)
compressible: Final = [m for i, m in enumerate(messages) if i not in protected_indices]
if not compressible:
return inputs
@ -847,12 +876,34 @@ class HeadroomGuardrail(CustomGuardrail):
# write-back and restructures it for nothing.
return inputs
restored: Final = (
_restore_compressed_text(compressible, result.messages)
if self.compress_latest_turn
else _restore_content_shapes(originals=compressible, returned=result.messages)
)
if restored is None:
return inputs
compressed: Final = _restore_protected_messages(
messages=messages,
compressed=_restore_content_shapes(originals=compressible, returned=result.messages),
compressed=restored,
protected_indices=protected_indices,
)
if self.compress_latest_turn and "input" in request_data:
from litellm.llms.openai.responses.guardrail_translation.handler import patch_response_request_fields
if (
patch_response_request_fields(
typed_request.get("input"),
typed_request.get("instructions"),
messages,
compressed,
preserve_text_metadata=True,
)
is None
):
return inputs
self.add_standard_logging_guardrail_information_to_request_data(
guardrail_json_response=result.stats,
request_data=request_data,

View file

@ -31,6 +31,15 @@ class HeadroomGuardrailConfigModel(GuardrailConfigModel[BaseModel]):
description="Inject the Headroom retrieval tool for hashes declared by the compression service.",
)
compress_latest_turn: bool = Field(
default=False,
description=(
"Allow compression of latest user and tool-result text, which can lose instructions or details. "
"System prompts, assistant messages, "
"cached prefixes and retrieved content remain protected. Unsafe rewrites retain the original input."
),
)
@staticmethod
def ui_friendly_name() -> str:
return "Headroom"

View file

@ -280,3 +280,14 @@ def test_embedding_scorer(respx_mock: respx.MockRouter) -> None:
{"role": "user", "content": "Fix auth"},
]
assert result["cache"] == {"message_1": recipe_context}
def test_relaxing_latest_turn_protection_retains_system_and_cached_prefix() -> None:
messages: Final = [
{"role": "system", "content": "instruction"},
{"role": "user", "content": "cached", "cache_control": {"type": "ephemeral"}},
{"role": "assistant", "content": "answer"},
{"role": "user", "content": "latest"},
]
assert get_protected_indices(messages, protect_latest_turn=False) == (0, 1)
assert set(get_protected_indices(messages)) == {0, 1, 2, 3}

View file

@ -7,9 +7,10 @@ with guardrail transformations, including tool calls.
import json
from collections.abc import Mapping
from typing import Any, Literal, Optional
from typing import Any, Final, Literal, Optional
import pytest
import respx
from litellm.integrations.custom_guardrail import CustomGuardrail
@ -2365,3 +2366,22 @@ class TestStreamingScanKey:
for released in (text_only, finished_tool_call):
ended = handler.released_stream_as_ended(released)
assert len(ended) == len(released) and all(a is b for a, b in zip(ended, released, strict=True))
@pytest.mark.asyncio
async def test_headroom_latest_user_rewrite_reaches_chat_request(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
import litellm
from litellm.proxy.guardrails.guardrail_hooks.headroom.headroom import HeadroomGuardrail
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base="https://headroom.example.com", compress_latest_turn=True)
route: Final = respx_mock.post("https://headroom.example.com/v1/compress").respond(
200, json={"messages": [{"role": "user", "content": "short"}]}
)
result: Final = await OpenAIChatCompletionsHandler().process_input_messages(
{"messages": [{"role": "user", "content": "long " * 100, "name": "reader"}]}, callback
)
assert route.called
assert result["messages"] == [{"role": "user", "content": "short", "name": "reader"}]

View file

@ -14,6 +14,8 @@ from unittest.mock import AsyncMock, MagicMock, patch
import logging
import pytest
import respx
import litellm
from fastapi import HTTPException
@ -4174,3 +4176,74 @@ class TestOpenAIResponsesHandlerStreamingScanKey:
assert [inputs.get("texts") for inputs in guardrail.seen_inputs] == [["hi"]], guardrail.seen_inputs
tool_calls = guardrail.seen_inputs[0].get("tool_calls") or []
assert [call["function"]["arguments"] for call in tool_calls] == ['{"city": "Paris"}'], tool_calls
@pytest.mark.asyncio
async def test_headroom_latest_turn_patches_native_tool_result_with_metadata(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
from litellm.proxy.guardrails.guardrail_hooks.headroom.headroom import HeadroomGuardrail
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base="https://headroom.example.com", compress_latest_turn=True)
items: Final = [
{"role": "user", "content": "Summarize this data"},
{"type": "function_call", "id": "fc_original", "call_id": "call_original", "name": "read", "arguments": "{}"},
{"type": "function_call_output", "call_id": "call_original", "output": "data " * 500, "status": "completed"},
]
route: Final = respx_mock.post("https://headroom.example.com/v1/compress").respond(200, json={"messages": [
{"role": "user", "content": "Summarize data"},
{"role": "tool", "tool_call_id": "call_original", "content": "short data"},
]})
result: Final = await OpenAIResponsesHandler().process_input_messages({"input": items}, callback)
assert route.called
assert result["input"] == [{**items[0], "content": "Summarize data"}, items[1], {**items[2], "output": "short data"}]
@pytest.mark.asyncio
@pytest.mark.parametrize("raw_input", ["long " * 100, [{"role": "user", "content": [
{"type": "input_text", "text": "long " * 100, "source": "first"},
{"type": "input_text", "text": "long " * 100, "source": "second"},
]}]])
async def test_headroom_latest_user_input_preserves_native_text_metadata(
raw_input: object, respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
from litellm.proxy.guardrails.guardrail_hooks.headroom.headroom import HeadroomGuardrail
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base="https://headroom.example.com", compress_latest_turn=True)
route: Final = respx_mock.post("https://headroom.example.com/v1/compress").respond(
200, json={"messages": [{"role": "user", "content": "short"}]}
)
result: Final = await OpenAIResponsesHandler().process_input_messages({"input": raw_input}, callback)
assert route.called
assert result["input"] == ([{"role": "user", "content": "short"}] if isinstance(raw_input, str) else [
{"role": "user", "content": [
{"type": "input_text", "text": "short", "source": "first"},
{"type": "input_text", "text": "", "source": "second"},
]},
])
@pytest.mark.asyncio
async def test_headroom_unsafe_native_mapping_preserves_request_and_tools(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
from litellm.proxy.guardrails.guardrail_hooks.headroom.headroom import HeadroomGuardrail
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base="https://headroom.example.com", compress_latest_turn=True)
items: Final = [
{"id": "fc_1", "type": "function_call", "call_id": "call_1", "name": "read", "arguments": "{}"},
{"role": "assistant", "content": "Reading now"},
{"type": "function_call_output", "call_id": "call_1", "output": "long " * 100},
]
original: Final = {"input": items}
route: Final = respx_mock.post("https://headroom.example.com/v1/compress").respond(200, json={
"messages": [{"role": "tool", "tool_call_id": "call_1", "content": "short"}],
"ccr_hashes": ["abcdefabcdef"],
})
result: Final = await OpenAIResponsesHandler().process_input_messages(original, callback)
assert route.called
assert result == {"input": items}
assert "tools" not in result

View file

@ -3075,3 +3075,139 @@ def test_unusable_timeout_falls_back_to_the_default(configured: float):
assert guardrail.timeout.read == 60.0
assert guardrail.timeout.connect == 5.0
@pytest.mark.asyncio
async def test_configured_latest_turn_compression_preserves_tool_linkage(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
from litellm.proxy.guardrails.guardrail_hooks.headroom import initialize_guardrail
from litellm.types.guardrails import LitellmParams
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
params: Final = LitellmParams.model_validate({
"guardrail": "headroom", "mode": "pre_call", "api_base": FAKE_API_BASE,
"compress_latest_turn": True,
})
callback: Final = initialize_guardrail(params, {"guardrail_name": "compress-latest", "litellm_params": params})
messages: Final = [
{"role": "user", "content": "Summarize this listing " * 100},
{"role": "assistant", "content": None, "tool_calls": [
{"id": "call_listing", "type": "function", "function": {"name": "list_files", "arguments": "{}"}},
]},
{"role": "tool", "tool_call_id": "call_listing", "content": "file.py\n" * 1000, "name": "list_files"},
]
compressed: Final = [
{"role": "user", "content": "Summarize the listing"},
{"role": "tool", "tool_call_id": "call_listing", "content": "1000 Python files"},
]
route: Final = respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(200, json={"messages": compressed})
try:
result: Final = await callback.apply_guardrail(
GenericGuardrailAPIInputs(structured_messages=messages), {}, "request"
)
assert result["structured_messages"] == [
{**messages[0], "content": compressed[0]["content"]}, messages[1],
{**messages[2], "content": compressed[1]["content"]},
]
assert json.loads(route.calls.last.request.content)["messages"] == [messages[0], messages[2]]
finally:
litellm.logging_callback_manager.remove_callback_from_list_by_object(litellm.callbacks, callback, require_self=False)
@pytest.mark.asyncio
@pytest.mark.parametrize("rewritten", [
[],
[{"role": "assistant", "content": "short"}],
[{"role": "user", "content": {"unexpected": "shape"}}],
[{"role": "user", "content": "short", "tool_call_id": "wrong"}],
])
async def test_latest_turn_unsafe_rewrite_keeps_original_input(
rewritten: list[dict[str, object]], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base=FAKE_API_BASE, compress_latest_turn=True)
original: Final = GenericGuardrailAPIInputs(structured_messages=[{"role": "user", "content": "long " * 100}])
route: Final = respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(200, json={"messages": rewritten})
result: Final = await callback.apply_guardrail(original, {}, "request")
assert route.called
assert result is original
@pytest.mark.asyncio
@pytest.mark.parametrize("enabled", [False, True])
async def test_latest_turn_preserves_cached_prefix_and_retrieval(
enabled: bool, respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base=FAKE_API_BASE, compress_latest_turn=enabled)
messages: Final = [
{"role": "system", "content": "System instruction"},
{"role": "user", "content": "Cached data", "cache_control": {"type": "ephemeral"}},
{"role": "assistant", "content": None, "tool_calls": [
{"id": "retrieve", "type": "function", "function": {"name": "headroom_retrieve", "arguments": "{}"}},
]},
{"role": "tool", "tool_call_id": "retrieve", "content": "Recovered original"},
{"role": "user", "content": "Latest instruction"},
]
with respx.mock(assert_all_called=False) as router:
route: Final = router.post(FAKE_API_BASE + "/v1/compress").respond(
200, json={"messages": [{"role": "user", "content": "Short instruction"}]}
)
result: Final = await callback.apply_guardrail(GenericGuardrailAPIInputs(structured_messages=messages), {}, "request")
assert route.called is enabled
assert result["structured_messages"] == [
*messages[:-1], {**messages[-1], "content": "Short instruction"} if enabled else messages[-1],
]
@pytest.mark.asyncio
async def test_latest_turn_retains_each_text_parts_metadata(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base=FAKE_API_BASE, compress_latest_turn=True)
messages: Final = [{"role": "user", "content": [
{"type": "text", "text": "long " * 100, "source": "first"},
{"type": "text", "text": "more " * 100, "source": "second"},
]}]
route: Final = respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(
200, json={"messages": [{"role": "user", "content": "short"}]}
)
result: Final = await callback.apply_guardrail(GenericGuardrailAPIInputs(structured_messages=messages), {}, "request")
assert route.called
assert result["structured_messages"] == [{"role": "user", "content": [
{"type": "text", "text": "short", "source": "first"},
{"type": "text", "text": "", "source": "second"},
]}]
@pytest.mark.asyncio
@pytest.mark.parametrize("content", [[], ["invalid"], [{"type": "image_url", "image_url": {"url": "https://example.com/image"}}]])
async def test_latest_turn_does_not_replace_non_text_content(
content: list[object], respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base=FAKE_API_BASE, compress_latest_turn=True)
original: Final = GenericGuardrailAPIInputs(structured_messages=[{"role": "user", "content": content}])
route: Final = respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(
200, json={"messages": [{"role": "user", "content": "short"}]}
)
result: Final = await callback.apply_guardrail(original, {}, "request")
assert route.called
assert result is original
@pytest.mark.asyncio
async def test_latest_turn_unchanged_text_retains_original_metadata(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base=FAKE_API_BASE, compress_latest_turn=True)
original: Final = [{"role": "user", "content": "Already short", "name": "reader"}]
route: Final = respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(
200, json={"messages": [{"role": "user", "content": "Already short"}]}
)
result: Final = await callback.apply_guardrail(GenericGuardrailAPIInputs(structured_messages=original), {}, "request")
assert route.called
assert result["structured_messages"] == original

View file

@ -36419,6 +36419,12 @@ export interface components {
* @description Client secret of the gateway's Entra app registration, used to perform the On-Behalf-Of exchange. Falls back to the AGENT365_CLIENT_SECRET environment variable.
*/
client_secret?: string | null;
/**
* Compress Latest Turn
* @description Allow compression of latest user and tool-result text, which can lose instructions or details. System prompts, assistant messages, cached prefixes and retrieved content remain protected. Unsafe rewrites retain the original input.
* @default false
*/
compress_latest_turn: boolean;
/**
* Confidence Threshold
* @description Only block or mask when detection confidence >= this value; below threshold, allow or log_only.