This commit is contained in:
joshua-berri 2026-10-20 20:24:08 -07:00 • committed by GitHub
commit 423f74af6d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 751 additions and 39 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

@ -32,6 +32,7 @@ import copy
import json
import time
import uuid
from collections import Counter
from collections.abc import Mapping, MutableSequence, Sequence
from dataclasses import dataclass
from itertools import accumulate, chain, repeat
@ -266,13 +267,21 @@ 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")))
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]
@ -472,6 +481,42 @@ def _input_item_provenance(
return item_for_message, tainted
_RESPONSE_INPUT_ITEMS_ADAPTER: Final = TypeAdapter(tuple[Mapping[str, object], ...])
def rewriteable_response_message_indices(
raw_input: object, instructions: object, original_messages: Sequence[object]
) -> frozenset[int]:
from litellm.proxy.guardrails.guardrail_hooks.content_text import is_rewriteable_text_content
input_items: Final = [{"role": "user", "content": raw_input}] if isinstance(raw_input, str) else raw_input
if not isinstance(input_items, list):
return frozenset()
try:
items: Final = _RESPONSE_INPUT_ITEMS_ADAPTER.validate_python(input_items)
except ValidationError:
return frozenset()
offset: Final = 1 if instructions else 0
provenance: Final = _input_item_provenance(items, tuple(original_messages)[offset:])
if provenance is None:
return frozenset()
item_for_message, tainted = provenance
text_rows: Final = tuple(
(message_index, item_index)
for message_index, item_index in item_for_message.items()
if stream_item_field(original_messages[message_index + offset], "role") != "assistant"
)
item_counts: Final = Counter(item_index for _, item_index in text_rows)
return frozenset(
message_index + offset
for message_index, item_index in text_rows
if message_index not in tainted
and item_counts[item_index] == 1
and (field := _item_rewrite_field(items[item_index])) is not None
and is_rewriteable_text_content(items[item_index].get(field))
)
class _RequestFields(NamedTuple):
input: tuple[object, ...]
instructions: str | None
@ -493,23 +538,30 @@ 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,
scoped_indices: Sequence[int] | None = None,
) -> _RequestFields | None:
if not isinstance(raw_input, list) or len(original_messages) != len(structured_messages):
input_items: Final = (
[{"role": "user", "content": raw_input}] if isinstance(raw_input, str) and preserve_text_metadata else raw_input
)
message_indices: Final = range(len(original_messages)) if scoped_indices is None else scoped_indices
if not isinstance(input_items, list) or len(message_indices) != len(structured_messages):
return None
offset: Final = 1 if instructions else 0
provenance: Final = _input_item_provenance(raw_input, tuple(original_messages)[offset:])
provenance: Final = _input_item_provenance(input_items, tuple(original_messages)[offset:])
if provenance is None:
return None
item_for_message, tainted = provenance
changed: Final = tuple(
(index, rewritten)
for index, (original, rewritten) in enumerate(zip(original_messages, structured_messages))
if original != rewritten
for index, rewritten in zip(message_indices, structured_messages)
if original_messages[index] != rewritten
)
instruction_rewrites: Final = tuple(rewritten for index, rewritten in changed if index < offset)
rewritten_instructions: Final = (
@ -526,8 +578,9 @@ def _patched_request_fields(
replacements: Final = MappingProxyType(
{
item_for_message[message_index]: _rewritten_input_item(
cast("Mapping[str, object]", raw_input[item_for_message[message_index]]), # cast-ok: checked Mappings
cast("Mapping[str, object]", input_items[item_for_message[message_index]]), # cast-ok: checked Mappings
rewritten,
preserve_text_metadata=preserve_text_metadata,
)
for message_index, rewritten in body_changes
}
@ -535,7 +588,7 @@ def _patched_request_fields(
if len(replacements) != len(body_changes) or any(item is None for item in replacements.values()):
return None
return _RequestFields(
input=tuple(replacements.get(index, item) for index, item in enumerate(raw_input)),
input=tuple(replacements.get(index, item) for index, item in enumerate(input_items)),
instructions=instructions_value,
)
@ -545,11 +598,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 +819,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

@ -17372,6 +17372,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,9 +12,11 @@ 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
@ -94,3 +96,27 @@ 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 is_rewriteable_text_content(content: object) -> bool:
if isinstance(content, str):
return True
if not isinstance(content, list) or len(content) != 1:
return False
try:
parts: Final = _TEXT_PARTS_ADAPTER.validate_python(content)
except ValidationError:
return False
return parts[0].get("type") in ("text", "input_text", "output_text") and isinstance(parts[0].get("text"), str)
def replace_text_content(content: object, text: str) -> object:
if not text.strip() or not is_rewriteable_text_content(content):
return None
if isinstance(content, str):
return text
parts: Final = _TEXT_PARTS_ADAPTER.validate_python(content)
return [{**parts[0], "text": text}]

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

@ -36,7 +36,9 @@ from litellm.proxy.guardrails.guardrail_hooks.content_text import (
assistant_text_from_response,
content_to_text,
is_all_text_parts,
is_rewriteable_text_content,
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 +143,34 @@ def _restore_content_shapes(
return restored
def _compressed_message_content(original: Mapping[str, object], rewritten: Mapping[str, object]) -> object:
content: Final = rewritten.get("content")
original_content: Final = original.get("content")
if content == original_content or (
is_all_text_parts(original_content) and content == content_to_text(original_content)
):
return original_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 +269,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 +290,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(
@ -269,14 +302,19 @@ def _protected_indices(
for index in group
)
| frozenset(index for index, message in enumerate(messages) if message.get("role") == "assistant")
| frozenset(
index
for index, message in enumerate(messages)
if compress_latest_turn and not is_rewriteable_text_content(message.get("content"))
)
)
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 +543,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 +557,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 +570,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 +720,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(
@ -773,11 +819,28 @@ class HeadroomGuardrail(CustomGuardrail):
return str(body)
def _record_skipped_rewrite(self, request_data: dict[str, object], start_time: float, end_time: float) -> None:
from litellm.proxy.common_utils.callback_utils import add_guardrail_to_applied_guardrails_header
self.add_standard_logging_guardrail_information_to_request_data(
guardrail_json_response={
"error": "Headroom rewrite could not be mapped safely",
"compression_applied": False,
},
request_data=request_data,
guardrail_status="guardrail_failed_to_respond",
guardrail_provider=HEADROOM_GUARDRAIL_PROVIDER,
start_time=start_time,
end_time=end_time,
duration=end_time - start_time,
)
add_guardrail_to_applied_guardrails_header(request_data=request_data, guardrail_name=self.guardrail_name)
@log_guardrail_information
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict,
request_data: dict[str, object],
input_type: Literal["request", "response"],
logging_obj: LiteLLMLoggingObj | None = None,
) -> GenericGuardrailAPIInputs:
@ -800,20 +863,44 @@ 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")
from litellm.llms.base_llm.guardrail_translation.utils import (
effective_skip_system_message_for_guardrail,
scoped_structured_message_indices,
)
from litellm.llms.openai.responses.guardrail_translation.handler import (
OpenAIResponsesHandler,
patch_response_request_fields,
rewriteable_response_message_indices,
)
native_request: Final = self.compress_latest_turn and "input" in request_data
full_messages: Final = (
OpenAIResponsesHandler().get_structured_messages(request_data) or [] if native_request else []
)
scoped_indices: Final = scoped_structured_message_indices(
full_messages,
scan_only_tool_results=False,
skip_system=effective_skip_system_message_for_guardrail(self),
skip_tool=False,
)
rewriteable: Final = (
rewriteable_response_message_indices(
request_data.get("input"), request_data.get("instructions"), full_messages
)
if native_request
else frozenset()
)
native_protected: Final = frozenset(
index for index, original_index in enumerate(scoped_indices) if original_index not in rewriteable
)
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)
| native_protected
)
compressible: Final = [m for i, m in enumerate(messages) if i not in protected_indices]
if not compressible:
return inputs
@ -847,12 +934,35 @@ 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:
self._record_skipped_rewrite(request_data, start_time, end_time)
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 (
native_request
and patch_response_request_fields(
typed_request.get("input"),
typed_request.get("instructions"),
full_messages,
compressed,
preserve_text_metadata=True,
scoped_indices=scoped_indices,
)
is None
):
self._record_skipped_rewrite(request_data, start_time, end_time)
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

@ -10,6 +10,8 @@ from typing import Any, Literal, Optional, Final
from unittest.mock import MagicMock, patch
import pytest
import respx
import litellm
from litellm.integrations.custom_guardrail import CustomGuardrail
@ -3043,3 +3045,22 @@ class TestAnthropicMessagesHandlerPostCallHookResponse:
native = {"type": "message", "role": "assistant", "content": [{"type": "text", "text": "hi"}]}
assert AnthropicMessagesHandler().post_call_hook_response(native) is native
@pytest.mark.asyncio
async def test_headroom_latest_multipart_turn_preserves_valid_text_blocks(
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)
messages: Final = [{"role": "user", "content": [
{"type": "text", "text": "Context\n\nparagraph " * 100},
{"type": "text", "text": "Summarize it"},
]}]
result: Final = await AnthropicMessagesHandler().process_input_messages(
{"model": "test-model", "max_tokens": 200, "messages": messages}, callback
)
assert not respx_mock.calls
assert result["messages"] == messages

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,193 @@ 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\n\nparagraph " * 100, "source": "first"},
]}]])
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\n\nsummary"}]}
)
result: Final = await OpenAIResponsesHandler().process_input_messages({"input": raw_input}, callback)
assert route.called
assert result["input"] == ([{"role": "user", "content": "short\n\nsummary"}] if isinstance(raw_input, str) else [
{"role": "user", "content": [
{"type": "input_text", "text": "short\n\nsummary", "source": "first"},
]},
])
@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
from litellm.litellm_core_utils.core_helpers import get_or_create_metadata_bucket
from litellm.proxy.spend_tracking.compression_savings import extract_compression_saved_tokens
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}
result: Final = await OpenAIResponsesHandler().process_input_messages(original, callback)
assert not respx_mock.calls
assert result["input"] == items
assert "tools" not in result
_, metadata = get_or_create_metadata_bucket(original)
entries: Final = metadata.get("standard_logging_guardrail_information", [])
assert entries == []
assert extract_compression_saved_tokens({"guardrail_information": entries}) == 0
@pytest.mark.asyncio
@pytest.mark.parametrize("skip_source", ["guardrail", "global"])
@pytest.mark.parametrize("include_system_item", [False, True])
async def test_headroom_skipped_system_messages_preserve_response_compression(
skip_source: str, include_system_item: bool, 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)
monkeypatch.setattr(litellm, "skip_system_message_in_guardrail", skip_source == "global")
callback: Final = HeadroomGuardrail(
api_base="https://headroom.example.com", compress_latest_turn=True,
)
callback.skip_system_message_in_guardrail = True if skip_source == "guardrail" else None
prefix: Final = [{"role": "system", "content": "Keep all identifiers"}] if include_system_item else []
items: Final = [*prefix,
{"role": "user", "content": "Read old data"},
{"type": "function_call", "call_id": "old_call", "name": "read", "arguments": "{}"},
{"type": "function_call_output", "call_id": "old_call", "output": "old " * 500},
{"role": "assistant", "content": "Done"},
{"role": "user", "content": "Read new data"},
{"type": "function_call", "call_id": "new_call", "name": "read", "arguments": "{}"},
{"type": "function_call_output", "call_id": "new_call", "output": "new " * 500},
]
route: Final = respx_mock.post("https://headroom.example.com/v1/compress").respond(200, json={"messages": [
{"role": "user", "content": "Read old"},
{"role": "tool", "tool_call_id": "old_call", "content": "old summary"},
{"role": "user", "content": "Read new"},
{"role": "tool", "tool_call_id": "new_call", "content": "new summary"},
]})
result: Final = await OpenAIResponsesHandler().process_input_messages(
{"instructions": "Be terse", "input": items}, callback
)
assert route.called
assert result["instructions"] == "Be terse"
assert result["input"] == [*prefix,
{"role": "user", "content": "Read old"}, items[len(prefix) + 1],
{"type": "function_call_output", "call_id": "old_call", "output": "old summary"},
{"role": "assistant", "content": "Done"},
{"role": "user", "content": "Read new"}, items[len(prefix) + 5],
{"type": "function_call_output", "call_id": "new_call", "output": "new summary"},
]
@pytest.mark.asyncio
@pytest.mark.parametrize("output", [
[{"type": "input_text", "text": "First\n\nparagraph", "source": "one"},
{"type": "input_text", "text": "Second", "source": "two"}],
[{"type": "input_text", "text": "Image context"},
{"type": "input_image", "image_url": "https://example.com/image.png"}],
])
async def test_headroom_native_tool_parts_remain_intact_while_latest_output_compresses(
output: list[dict[str, 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)
items: Final = [
{"type": "function_call", "call_id": "old", "name": "read", "arguments": "{}"},
{"type": "function_call_output", "call_id": "old", "output": output},
{"role": "user", "content": "Read new data"},
{"type": "function_call", "call_id": "new", "name": "read", "arguments": "{}"},
{"type": "function_call_output", "call_id": "new", "output": "new " * 500},
]
route: Final = respx_mock.post("https://headroom.example.com/v1/compress").respond(200, json={"messages": [
{"role": "user", "content": "Read new data"},
{"role": "tool", "tool_call_id": "new", "content": "new summary"},
]})
result: Final = await OpenAIResponsesHandler().process_input_messages({"input": items}, callback)
assert json.loads(route.calls.last.request.content)["messages"] == [
{"role": "user", "content": "Read new data"},
{"role": "tool", "tool_call_id": "new", "content": "new " * 500},
]
assert result["input"] == [*items[:-1], {**items[-1], "output": "new summary"}]
@pytest.mark.parametrize("raw_input", [None, 42, ["invalid item"], [{1: "invalid key"}]])
def test_native_compression_excludes_invalid_input_shapes(raw_input: object) -> None:
from litellm.llms.openai.responses.guardrail_translation.handler import rewriteable_response_message_indices
assert rewriteable_response_message_indices(raw_input, None, []) == frozenset()
@pytest.mark.asyncio
async def test_headroom_cached_tool_definition_preserves_latest_output_compression(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
from litellm.proxy.guardrails.guardrail_hooks.headroom.headroom import HeadroomGuardrail
from litellm.responses.litellm_completion_transformation.transformation import TOOL_CALLS_CACHE
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
call_id: Final = "call_headroom_cached_definition"
TOOL_CALLS_CACHE.set_cache(key=call_id, value={
"id": call_id, "type": "function", "function": {"name": "read", "arguments": "{}"},
})
callback: Final = HeadroomGuardrail(api_base="https://headroom.example.com", compress_latest_turn=True)
items: Final = [
{"role": "user", "content": "Read data"},
{"type": "function_call_output", "call_id": call_id, "output": "long " * 500},
]
route: Final = respx_mock.post("https://headroom.example.com/v1/compress").respond(200, json={"messages": [
{"role": "user", "content": "Read data"},
{"role": "tool", "tool_call_id": call_id, "content": "short summary"},
]})
try:
result: Final = await OpenAIResponsesHandler().process_input_messages({"input": items}, callback)
assert route.called
assert json.loads(route.calls.last.request.content)["messages"] == [
{"role": "user", "content": "Read data"},
{"role": "tool", "tool_call_id": call_id, "content": "long " * 500},
]
assert result["input"] == [items[0], {**items[1], "output": "short summary"}]
finally:
TOOL_CALLS_CACHE.delete_cache(key=call_id)

View file

@ -3075,3 +3075,246 @@ 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": ""}],
[{"role": "user", "content": " "}],
[{"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, "ccr_hashes": [CCR_HASH]})
request_data: Final[dict[str, object]] = {}
result: Final = await callback.apply_guardrail(original, request_data, "request")
assert route.called
assert result is original
assert "tools" not in result
entries: Final = _recorded_guardrail_entries(request_data)
assert len(entries) == 1
assert entries[0]["guardrail_status"] == "guardrail_failed_to_respond"
assert entries[0]["duration"] >= 0
assert entries[0]["end_time"] >= entries[0]["start_time"]
assert extract_compression_saved_tokens({"guardrail_information": entries}) == 0
@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_single_text_block_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\n\nparagraph " * 100, "source": "first"},
]}]
route: Final = respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(
200, json={"messages": [{"role": "user", "content": "short\n\nsummary"}]}
)
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\n\nsummary", "source": "first"},
]}]
@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}])
result: Final = await callback.apply_guardrail(original, {}, "request")
assert not respx_mock.calls
assert result is original
@pytest.mark.asyncio
@pytest.mark.parametrize("content", ["Already short", [
{"type": "text", "text": "Already", "source": "first"},
{"type": "text", "text": "short", "source": "second"},
]])
async def test_latest_turn_unchanged_text_retains_original_metadata(
content: 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 = [{"role": "user", "content": content, "name": "reader"}]
if isinstance(content, str):
respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(
200, json={"messages": [
{"role": "user", "content": "Already short" if isinstance(content, str) else "Already\n\nshort"}
]}
)
result: Final = await callback.apply_guardrail(
GenericGuardrailAPIInputs(structured_messages=original), {}, "request"
)
assert bool(respx_mock.calls) is isinstance(content, str)
assert result["structured_messages"] == original
@pytest.mark.asyncio
@pytest.mark.parametrize("surface", ["chat", "messages", "responses"])
async def test_ambiguous_history_does_not_prevent_latest_tool_compression(
surface: str, respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
from litellm.llms.anthropic.chat.guardrail_translation.handler import AnthropicMessagesHandler
from litellm.llms.openai.chat.guardrail_translation.handler import OpenAIChatCompletionsHandler
from litellm.llms.openai.responses.guardrail_translation.handler import OpenAIResponsesHandler
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
callback: Final = HeadroomGuardrail(api_base=FAKE_API_BASE, compress_latest_turn=True)
part_type: Final = "input_text" if surface == "responses" else "text"
history: Final = {"role": "user", "content": [
{"type": part_type, "text": "First paragraph\n\nSecond paragraph", **({} if surface == "messages" else {"source": "first"})},
{"type": part_type, "text": "Other block", **({} if surface == "messages" else {"source": "second"})},
]}
tool_text: Final = "file.py\n" * 1000
prefix: Final = [history, {"role": "assistant", "content": "Read the history"}]
chat_rows: Final = [*prefix, {"role": "user", "content": "List files"},
{"role": "assistant", "content": None, "tool_calls": [
{"id": "call_files", "type": "function", "function": {"name": "list_files", "arguments": "{}"}},
]}, {"role": "tool", "tool_call_id": "call_files", "content": tool_text},
]
native_items: Final = [*prefix, {"role": "user", "content": "List files"},
{"type": "function_call", "call_id": "call_files", "name": "list_files", "arguments": "{}"},
{"type": "function_call_output", "call_id": "call_files", "output": tool_text},
]
anthropic_rows: Final = [*prefix, {"role": "user", "content": "List files"},
{"role": "assistant", "content": [{"type": "tool_use", "id": "call_files", "name": "list_files", "input": {}}]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call_files", "content": tool_text}]},
]
request: Final = {"input": native_items} if surface == "responses" else {
"messages": anthropic_rows if surface == "messages" else chat_rows
}
route: Final = respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(200, json={
"messages": [{"role": "user", "content": "List files"},
{"role": "tool", "tool_call_id": "call_files", "content": "file summary"}],
"tokens_before": 1000, "tokens_after": 10, "ccr_hashes": [CCR_HASH],
})
handler: Final = (OpenAIResponsesHandler() if surface == "responses" else
AnthropicMessagesHandler() if surface == "messages" else OpenAIChatCompletionsHandler())
result: Final = await handler.process_input_messages({**request, "model": "test-model", "max_tokens": 100}, callback)
assert json.loads(route.calls.last.request.content)["messages"] == [
{"role": "user", "content": "List files"},
{"role": "tool", "tool_call_id": "call_files", "content": tool_text},
]
if surface == "responses":
assert result["input"] == [*native_items[:-1], {**native_items[-1], "output": "file summary"}]
elif surface == "chat":
assert result["messages"] == [*chat_rows[:-1], {**chat_rows[-1], "content": "file summary"}]
else:
assert result["messages"][0] == history
assert result["messages"][-1]["content"] == [
{"type": "tool_result", "tool_use_id": "call_files", "content": "file summary"}
]
assert extract_compression_saved_tokens({"guardrail_information": _recorded_guardrail_entries(result)}) == 990
assert has_headroom_retrieve_tool(result.get("tools"))
@pytest.mark.asyncio
async def test_native_scope_mismatch_discards_rewrite_and_retrieval(
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}])
request: Final = {"input": [
{"role": "user", "content": "long " * 100}, {"role": "user", "content": "Additional input"},
]}
route: Final = respx_mock.post(FAKE_API_BASE + "/v1/compress").respond(200, json={
"messages": [{"role": "user", "content": "short"}],
"ccr_hashes": [CCR_HASH], "tokens_before": 1000, "tokens_after": 10,
})
result: Final = await callback.apply_guardrail(original, request, "request")
assert route.called
assert result is original
assert "tools" not in result
entries: Final = _recorded_guardrail_entries(request)
assert len(entries) == 1
assert entries[0]["guardrail_status"] == "guardrail_failed_to_respond"
assert entries[0]["duration"] >= 0
assert extract_compression_saved_tokens({"guardrail_information": entries}) == 0

View file

@ -36572,6 +36572,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.