74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
"""Tests for prompt-only trajectory compaction."""
|
|
|
|
from swe_data_processing.evidence import (
|
|
BOUNDARY_BLOCK_TURNS,
|
|
build_trajectory_blocks,
|
|
compact_text,
|
|
compact_trajectory,
|
|
)
|
|
|
|
|
|
def test_compact_text_preserves_short_values() -> None:
|
|
"""Short evidence must remain byte-for-byte identical."""
|
|
|
|
assert compact_text("short evidence", 100) == "short evidence"
|
|
|
|
|
|
def test_compact_text_hashes_long_values() -> None:
|
|
"""Long evidence must expose its size and digest for auditability."""
|
|
|
|
result = compact_text("a" * 1_000, 200)
|
|
assert len(result) == 200
|
|
assert "COMPACTED original_chars=1000 sha256=" in result
|
|
|
|
|
|
def test_compact_trajectory_prioritizes_test_observations() -> None:
|
|
"""Important test output receives a larger preview than ordinary output."""
|
|
|
|
long_content = "x" * 3_000
|
|
trajectory = [
|
|
{"turn_id": 1, "role": "user", "content": "issue"},
|
|
{"turn_id": 2, "role": "assistant", "content": "", "tool_calls": []},
|
|
{"turn_id": 3, "role": "tool", "content": long_content},
|
|
{"turn_id": 4, "role": "assistant", "content": "done"},
|
|
]
|
|
signals = {
|
|
"stateful_turns": [],
|
|
"test_events": [{"command_turn": 2, "output_turn": 3}],
|
|
"malformed_tool_turns": [],
|
|
"unknown_tool_turns": [],
|
|
}
|
|
compacted, metadata = compact_trajectory(trajectory, signals)
|
|
assert compacted[2]["content"] == long_content
|
|
assert 3 in metadata["important_turn_ids"]
|
|
|
|
|
|
def test_boundary_blocks_preserve_absolute_turns_without_overlap() -> None:
|
|
trajectory = [
|
|
{"turn_id": turn, "role": "assistant", "content": f"turn {turn}"}
|
|
for turn in range(1, BOUNDARY_BLOCK_TURNS + 3)
|
|
]
|
|
blocks, metadata = build_trajectory_blocks(trajectory, {})
|
|
assert len(blocks) == 2
|
|
assert blocks[0]["start_turn"] == 1
|
|
assert blocks[0]["end_turn"] == BOUNDARY_BLOCK_TURNS
|
|
assert blocks[1]["start_turn"] == BOUNDARY_BLOCK_TURNS + 1
|
|
assert blocks[1]["end_turn"] == BOUNDARY_BLOCK_TURNS + 2
|
|
turn_ids = [
|
|
message["turn_id"] for block in blocks for message in block["messages"]
|
|
]
|
|
assert turn_ids == list(range(1, BOUNDARY_BLOCK_TURNS + 3))
|
|
assert metadata["block_count"] == 2
|
|
|
|
|
|
def test_boundary_block_keeps_immediate_tool_result_with_assistant() -> None:
|
|
trajectory = [
|
|
{"turn_id": turn, "role": "user", "content": f"turn {turn}"}
|
|
for turn in range(1, BOUNDARY_BLOCK_TURNS + 2)
|
|
]
|
|
trajectory[BOUNDARY_BLOCK_TURNS - 1]["role"] = "assistant"
|
|
trajectory[BOUNDARY_BLOCK_TURNS]["role"] = "tool"
|
|
blocks, _ = build_trajectory_blocks(trajectory, {})
|
|
assert blocks[0]["end_turn"] == BOUNDARY_BLOCK_TURNS + 1
|
|
assert len(blocks) == 1
|