313 lines
10 KiB
Python
313 lines
10 KiB
Python
"""Regression tests for blank user echoes displacing actionable compaction state."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from agent.context_compressor import (
|
|
COMPRESSED_SUMMARY_METADATA_KEY,
|
|
SUMMARY_PREFIX,
|
|
ContextCompressor,
|
|
)
|
|
|
|
|
|
@pytest.fixture()
|
|
def compressor() -> ContextCompressor:
|
|
with patch(
|
|
"agent.context_compressor.get_model_context_length",
|
|
return_value=100_000,
|
|
):
|
|
instance = ContextCompressor(
|
|
model="test/model",
|
|
threshold_percent=0.85,
|
|
protect_first_n=2,
|
|
protect_last_n=2,
|
|
quiet_mode=True,
|
|
)
|
|
instance.tail_token_budget = 10
|
|
return instance
|
|
|
|
|
|
def _append_tool_run(messages: list[dict], prefix: str, count: int = 6) -> None:
|
|
for index in range(count):
|
|
call_id = f"{prefix}-{index}"
|
|
messages.extend(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": call_id,
|
|
"function": {"name": "read_file", "arguments": "{}"},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": call_id,
|
|
"content": "x" * 400,
|
|
},
|
|
]
|
|
)
|
|
|
|
|
|
def _compress(compressor: ContextCompressor, messages: list[dict]) -> list[dict]:
|
|
with patch.object(
|
|
compressor,
|
|
"_generate_summary",
|
|
return_value=f"{SUMMARY_PREFIX}\nsummary of older work",
|
|
):
|
|
return compressor.compress(messages, current_tokens=90_000)
|
|
|
|
|
|
def _assert_no_adjacent_user_roles(messages: list[dict]) -> None:
|
|
for previous, current in zip(messages, messages[1:]):
|
|
assert (previous.get("role"), current.get("role")) != ("user", "user")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"blank",
|
|
[
|
|
"",
|
|
" \n\t",
|
|
None,
|
|
[],
|
|
[{"type": "text", "text": " "}],
|
|
[{"type": "input_text", "text": " "}],
|
|
],
|
|
)
|
|
def test_blank_echo_does_not_displace_async_completion(compressor, blank):
|
|
completion = "[ASYNC DELEGATION BATCH COMPLETE — deleg_current]\nnew result"
|
|
messages = [
|
|
{"role": "system", "content": "sys"},
|
|
{"role": "user", "content": "old request"},
|
|
{"role": "assistant", "content": "old reply"},
|
|
{"role": "user", "content": completion},
|
|
{"role": "user", "content": blank},
|
|
{"role": "assistant", "content": "working from the completion"},
|
|
]
|
|
|
|
assert compressor._find_last_user_message_idx(messages, head_end=1) == 3
|
|
|
|
|
|
def test_leading_blank_without_actionable_user_is_not_removed(compressor):
|
|
messages = [
|
|
{"role": "user", "content": ""},
|
|
{"role": "assistant", "content": "visible reply"},
|
|
]
|
|
|
|
assert compressor._find_last_user_message_idx(messages, 0) == -1
|
|
assert compressor._blank_echo_indices_after(messages, -1) == set()
|
|
|
|
|
|
def test_image_only_user_turn_survives_compaction(compressor):
|
|
image_content = [
|
|
{
|
|
"type": "image_url",
|
|
"image_url": {"url": "data:image/png;base64,AA=="},
|
|
}
|
|
]
|
|
messages: list[dict] = [
|
|
{"role": "system", "content": "sys"},
|
|
{"role": "user", "content": "old request"},
|
|
{"role": "assistant", "content": "old reply"},
|
|
]
|
|
messages += [
|
|
{"role": "user", "content": f"older question {index}"}
|
|
if index % 2 == 0
|
|
else {"role": "assistant", "content": f"older reply {index}"}
|
|
for index in range(6)
|
|
]
|
|
messages += [
|
|
{"role": "user", "content": image_content},
|
|
{"role": "user", "content": ""},
|
|
{"role": "assistant", "content": "analyzing the image"},
|
|
]
|
|
_append_tool_run(messages, "image")
|
|
|
|
result = _compress(compressor, messages)
|
|
|
|
assert any(message.get("content") == image_content for message in result)
|
|
assert all(not compressor._is_blank_user_turn(message) for message in result)
|
|
_assert_no_adjacent_user_roles(result)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"payload",
|
|
[
|
|
[{"type": "audio", "source": {"data": "AA=="}}],
|
|
[{"type": "input_audio", "input_audio": {"data": "AA=="}}],
|
|
[{"type": "future_input", "payload": {"value": 7}}],
|
|
],
|
|
ids=["audio", "input-audio", "unknown-structured"],
|
|
)
|
|
def test_structured_non_text_user_turn_survives_compaction(compressor, payload):
|
|
messages: list[dict] = [
|
|
{"role": "system", "content": "sys"},
|
|
{"role": "user", "content": "old request"},
|
|
{"role": "assistant", "content": "old reply"},
|
|
]
|
|
messages += [
|
|
{"role": "user", "content": f"older question {index}"}
|
|
if index % 2 == 0
|
|
else {"role": "assistant", "content": f"older reply {index}"}
|
|
for index in range(6)
|
|
]
|
|
messages += [
|
|
{"role": "user", "content": payload},
|
|
{"role": "user", "content": ""},
|
|
{"role": "assistant", "content": "processing structured input"},
|
|
]
|
|
_append_tool_run(messages, "structured")
|
|
|
|
result = _compress(compressor, messages)
|
|
|
|
assert any(message.get("content") == payload for message in result)
|
|
assert all(not compressor._is_blank_user_turn(message) for message in result)
|
|
_assert_no_adjacent_user_roles(result)
|
|
|
|
|
|
def test_completion_survives_compaction_verbatim_after_blank_echo(compressor):
|
|
completion = (
|
|
"[ASYNC DELEGATION BATCH COMPLETE — deleg_current]\n"
|
|
"The newest result that must remain actionable."
|
|
)
|
|
messages: list[dict] = [
|
|
{"role": "system", "content": "sys"},
|
|
{"role": "user", "content": "initial request"},
|
|
{"role": "assistant", "content": "initial reply"},
|
|
]
|
|
messages += [
|
|
{"role": "user", "content": f"older question {index}"}
|
|
if index % 2 == 0
|
|
else {"role": "assistant", "content": f"older reply {index}"}
|
|
for index in range(6)
|
|
]
|
|
messages += [
|
|
{"role": "user", "content": completion},
|
|
{"role": "user", "content": " \n"},
|
|
{"role": "assistant", "content": "working from the completion"},
|
|
]
|
|
_append_tool_run(messages, "tail")
|
|
|
|
result = _compress(compressor, messages)
|
|
|
|
completion_rows = [message for message in result if message.get("content") == completion]
|
|
assert len(completion_rows) == 1
|
|
assert not completion_rows[0].get(COMPRESSED_SUMMARY_METADATA_KEY)
|
|
summary_rows = [
|
|
message for message in result if message.get(COMPRESSED_SUMMARY_METADATA_KEY)
|
|
]
|
|
assert len(summary_rows) == 1
|
|
assert summary_rows[0].get("role") == "user"
|
|
assert all(not compressor._is_blank_user_turn(message) for message in result)
|
|
_assert_no_adjacent_user_roles(result)
|
|
|
|
second_result = _compress(compressor, result)
|
|
second_completion_rows = [
|
|
message for message in second_result if message.get("content") == completion
|
|
]
|
|
assert len(second_completion_rows) == 1
|
|
assert not second_completion_rows[0].get(COMPRESSED_SUMMARY_METADATA_KEY)
|
|
|
|
|
|
def test_completion_at_compress_start_survives_when_blank_echo_is_compress_end(
|
|
compressor,
|
|
):
|
|
completion = "latest actionable completion at the compression boundary"
|
|
messages = [
|
|
{"role": "system", "content": "sys"},
|
|
{"role": "user", "content": "initial request"},
|
|
{"role": "assistant", "content": "initial reply"},
|
|
{"role": "user", "content": completion},
|
|
{"role": "user", "content": ""},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": "boundary-call",
|
|
"function": {"name": "read_file", "arguments": "{}"},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": "boundary-call", "content": "result"},
|
|
{"role": "assistant", "content": "working from the completion"},
|
|
]
|
|
|
|
with (
|
|
patch.object(compressor, "_protect_head_size", return_value=3),
|
|
patch.object(compressor, "_find_tail_cut_by_tokens", return_value=3),
|
|
patch.object(compressor, "_generate_summary") as generate_summary,
|
|
):
|
|
result = compressor.compress(messages, current_tokens=90_000)
|
|
|
|
completion_rows = [message for message in result if message.get("content") == completion]
|
|
assert len(completion_rows) == 1
|
|
assert not completion_rows[0].get(COMPRESSED_SUMMARY_METADATA_KEY)
|
|
assert not any(
|
|
message.get(COMPRESSED_SUMMARY_METADATA_KEY) for message in result
|
|
)
|
|
assert len(result) == len(messages) - 1
|
|
assert compressor.compression_count == 0
|
|
assert compressor._last_compression_savings_pct == 0.0
|
|
generate_summary.assert_not_called()
|
|
assert any(message.get("tool_call_id") == "boundary-call" for message in result)
|
|
assert result[-1].get("content") == "working from the completion"
|
|
assert [message.get("role") for message in result] == [
|
|
"system",
|
|
"user",
|
|
"assistant",
|
|
"user",
|
|
"assistant",
|
|
"tool",
|
|
"assistant",
|
|
]
|
|
_assert_no_adjacent_user_roles(result)
|
|
|
|
|
|
def test_tool_call_head_compacts_without_rewriting_event(compressor):
|
|
completion = "latest actionable completion"
|
|
messages: list[dict] = [
|
|
{"role": "user", "content": "initial request"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": "head-call",
|
|
"function": {"name": "read_file", "arguments": "{}"},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": "head-call", "content": "head result"},
|
|
]
|
|
messages += [
|
|
{"role": "user", "content": f"older question {index}"}
|
|
if index % 2 == 0
|
|
else {"role": "assistant", "content": f"older reply {index}"}
|
|
for index in range(6)
|
|
]
|
|
messages += [
|
|
{"role": "user", "content": completion},
|
|
{"role": "user", "content": ""},
|
|
{"role": "assistant", "content": "working"},
|
|
]
|
|
_append_tool_run(messages, "tail")
|
|
|
|
result = _compress(compressor, messages)
|
|
|
|
assert compressor._last_compress_aborted is False
|
|
assert any(message.get("content") == completion for message in result)
|
|
head = next(
|
|
message
|
|
for message in result
|
|
if any(call.get("id") == "head-call" for call in message.get("tool_calls", []))
|
|
)
|
|
assert not head.get(COMPRESSED_SUMMARY_METADATA_KEY)
|
|
assert any(message.get("tool_call_id") == "head-call" for message in result)
|
|
_assert_no_adjacent_user_roles(result)
|