scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-17 04:34:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 173

test(guardrails): cover generic agent multi-turn incremental scan

Browse source 
Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>
This commit is contained in:
Devin AI 2026-07-16 21:17:29 +00:00
parent 05dcf54efd
commit 68340a6c05
1 changed files with 56 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
tests/test_litellm/proxy/guardrails/guardrail_hooks/test_bedrock_guardrails.py
View file

@ -3399,6 +3399,62 @@ class TestBedrockOnlyScanNewMessages:
assert first_scanned is not None
assert [m["content"] for m in first_scanned] == ["my ssn is 123-45-6789"]
@pytest.mark.asyncio
async def test_generic_agent_multi_turn_scans_only_new_each_turn(self):
"""A generic agent (not Claude Code) opts in by propagating a session id.
Agent frameworks on the OpenAI SDK carry the session through the request
body (metadata.session_id here), not the x-claude-code-session-id header.
Across a growing multi-turn conversation every turn after the first must
send Bedrock only the newly appended segments, never the whole context.
"""
guardrail = self._guardrail()
session = {"metadata": {"session_id": "agent-multi-turn"}}
with patch.object(guardrail, "make_bedrock_api_request", new_callable=AsyncMock) as mock_api:
mock_api.return_value = {"action": "NONE", "output": [], "outputs": []}
await guardrail.apply_guardrail(
inputs={"texts": ["system prompt", "turn 1 question"]},
request_data=session,
input_type="request",
)
assert [m["content"] for m in mock_api.call_args.kwargs["messages"]] == [
"system prompt",
"turn 1 question",
]
mock_api.reset_mock()
await guardrail.apply_guardrail(
inputs={"texts": ["system prompt", "turn 1 question", "turn 1 answer", "turn 2 question"]},
request_data=session,
input_type="request",
)
assert [m["content"] for m in mock_api.call_args.kwargs["messages"]] == [
"turn 1 answer",
"turn 2 question",
]
mock_api.reset_mock()
await guardrail.apply_guardrail(
inputs={
"texts": [
"system prompt",
"turn 1 question",
"turn 1 answer",
"turn 2 question",
"turn 2 answer",
"turn 3 question",
]
},
request_data=session,
input_type="request",
)
assert [m["content"] for m in mock_api.call_args.kwargs["messages"]] == [
"turn 2 answer",
"turn 3 question",
]
@pytest.mark.asyncio
async def test_blocked_turn_is_rescanned_on_retry(self):
guardrail = self._guardrail()