wehub-resource-sync
254 lines
9.0 KiB
Python
254 lines
9.0 KiB
Python
"""Tests for the compaction compaction public entry point.
|
|
|
|
: the dispatcher env-var, legacy implementation, and shadow comparator
|
|
have been deleted. ``compact_context`` delegates directly to
|
|
``compact_context_new``. These tests cover the public contract and the
|
|
turn-boundary cut behavior introduced in compaction.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from opensquilla.session.compaction import (
|
|
CompactionConfig,
|
|
CompactionRequest,
|
|
CompactionResult,
|
|
compact_context,
|
|
compact_context_new,
|
|
)
|
|
|
|
|
|
def _make_request(
|
|
entries: list[dict] | None = None,
|
|
window: int = 8192,
|
|
) -> CompactionRequest:
|
|
if entries is None:
|
|
entries = [
|
|
{"role": "user", "content": "hello", "token_count": 5},
|
|
{"role": "assistant", "content": "world", "token_count": 5},
|
|
]
|
|
return CompactionRequest(
|
|
session_id="dispatcher-test",
|
|
entries=entries,
|
|
context_window_tokens=window,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compact_context delegates to compact_context_new
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compact_context_returns_compaction_result():
|
|
"""compact_context returns a CompactionResult without raising."""
|
|
result = await compact_context(_make_request())
|
|
assert isinstance(result, CompactionResult)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compact_context_delegates_to_new(monkeypatch):
|
|
"""compact_context calls compact_context_new (not any legacy path)."""
|
|
import opensquilla.session.compaction as compaction_mod
|
|
|
|
calls = []
|
|
|
|
async def spy_new(request):
|
|
calls.append(request)
|
|
return CompactionResult(
|
|
summary="spy",
|
|
kept_entries=[],
|
|
removed_count=0,
|
|
chunks_processed=0,
|
|
summary_source="skipped",
|
|
tokens_before=0,
|
|
tokens_after=0,
|
|
remaining_budget_tokens=8192,
|
|
)
|
|
|
|
monkeypatch.setattr(compaction_mod, "compact_context_new", spy_new)
|
|
result = await compact_context(_make_request())
|
|
assert result.summary == "spy"
|
|
assert len(calls) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_compact_context_noop_when_within_budget():
|
|
"""When total tokens fit in the window, nothing is removed."""
|
|
result = await compact_context(_make_request(window=8192))
|
|
assert result.removed_count == 0
|
|
assert result.summary_source == "skipped"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# compact_context_new: turn-boundary cut
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_avoids_mid_turn_cut():
|
|
"""Turn-boundary cut must not split an assistant tool_call from its result."""
|
|
# Build entries that would be split mid-turn by a pure token-budget cut.
|
|
# Window = 30 tokens; keep_budget = 15.
|
|
# Entries: u=5 a_tool=5 tool_result=5 u=5 a=5 (total=25)
|
|
# A raw token-budget cut from the end would try to keep 15 tokens:
|
|
# a_final(5) + u_2(5) = 10 < 15, then tool_result(5) = 15 — keep 3 entries
|
|
# last_removed = a_tool → mid_turn!
|
|
# The new impl should walk back one more step so last_removed = u_1 (turn_boundary).
|
|
entries = [
|
|
{"role": "user", "content": "q1", "token_count": 5},
|
|
{
|
|
"role": "assistant",
|
|
"content": "[tool_call:read_file({\"path\": \"x\"})]",
|
|
"token_count": 5,
|
|
},
|
|
{"role": "tool", "content": "[tool_result:read_file] contents", "token_count": 5},
|
|
{"role": "user", "content": "q2", "token_count": 5},
|
|
{"role": "assistant", "content": "answer", "token_count": 5},
|
|
]
|
|
# Use a small window to force compaction; no API key so fallback summary is used.
|
|
request = CompactionRequest(
|
|
session_id="boundary-test",
|
|
entries=entries,
|
|
context_window_tokens=22, # total=25 > 22*safety(1.2)=26.4 — actually under; use tighter
|
|
config=CompactionConfig(safety_margin=1.0),
|
|
)
|
|
result = await compact_context_new(request)
|
|
|
|
if result.removed_count > 0:
|
|
# The cut should NOT land with last_removed = assistant tool_call
|
|
# and first_kept = tool_result.
|
|
removed = entries[: len(entries) - len(result.kept_entries)]
|
|
kept = result.kept_entries
|
|
if removed and kept:
|
|
last_removed_role = removed[-1].get("role")
|
|
last_removed_content = str(removed[-1].get("content") or "")
|
|
first_kept_role = kept[0].get("role")
|
|
is_mid_turn = (
|
|
last_removed_role == "assistant"
|
|
and "[tool_call:" in last_removed_content
|
|
and first_kept_role == "tool"
|
|
)
|
|
assert not is_mid_turn, (
|
|
f"Cut landed mid-turn: last_removed={removed[-1]}, first_kept={kept[0]}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_avoids_mid_turn_cut_for_agent_flattened_tool_blocks():
|
|
"""Turn-boundary cut must match the Agent's flattened tool-use entries."""
|
|
entries = [
|
|
{"role": "user", "content": "old context", "token_count": 100},
|
|
{"role": "user", "content": "q1", "token_count": 100},
|
|
{"role": "assistant", "content": "[Used tool: read_file]", "token_count": 5},
|
|
{
|
|
"role": "user",
|
|
"content": "[Tool result (toolu_1): file contents]",
|
|
"token_count": 5,
|
|
},
|
|
{"role": "user", "content": "q2", "token_count": 5},
|
|
{"role": "assistant", "content": "answer", "token_count": 5},
|
|
]
|
|
request = CompactionRequest(
|
|
session_id="agent-flattened-boundary-test",
|
|
entries=entries,
|
|
context_window_tokens=100,
|
|
config=CompactionConfig(safety_margin=1.0),
|
|
)
|
|
result = await compact_context_new(request)
|
|
|
|
assert result.removed_count > 0
|
|
removed = entries[: len(entries) - len(result.kept_entries)]
|
|
kept = result.kept_entries
|
|
assert removed[-1]["content"] != "[Used tool: read_file]"
|
|
assert kept[0]["content"] == "[Used tool: read_file]"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_skips_when_only_cut_would_orphan_tool_result():
|
|
"""If no clean boundary exists, compaction must not split tool state."""
|
|
entries = [
|
|
{
|
|
"role": "assistant",
|
|
"content": "calling tool",
|
|
"tool_calls": [{"id": "call_1", "type": "function"}],
|
|
"token_count": 4,
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"content": "tool result",
|
|
"tool_call_id": "call_1",
|
|
"token_count": 4,
|
|
},
|
|
{"role": "user", "content": "q2", "token_count": 3},
|
|
{"role": "assistant", "content": "answer", "token_count": 3},
|
|
]
|
|
request = CompactionRequest(
|
|
session_id="boundary-start-test",
|
|
entries=entries,
|
|
context_window_tokens=26,
|
|
config=CompactionConfig(safety_margin=1.0),
|
|
)
|
|
|
|
result = await compact_context_new(request)
|
|
|
|
assert result.removed_count == 0
|
|
assert result.summary_source == "skipped"
|
|
assert result.kept_entries == entries
|
|
assert result.skip_reason == "no_safe_turn_boundary"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_prev_summary_prefix_injected():
|
|
"""When custom_instructions carries __prev_summary__: the merged summary is prefixed."""
|
|
entries = [
|
|
{"role": "user", "content": "a " * 500, "token_count": 500},
|
|
{"role": "assistant", "content": "b " * 500, "token_count": 500},
|
|
]
|
|
request = CompactionRequest(
|
|
session_id="prev-summary-test",
|
|
entries=entries,
|
|
context_window_tokens=100,
|
|
config=CompactionConfig(safety_margin=1.0),
|
|
custom_instructions="__prev_summary__:prior context here\nnormal instructions",
|
|
)
|
|
result = await compact_context_new(request)
|
|
if result.removed_count > 0:
|
|
assert "[Previous context]" in result.summary
|
|
assert "prior context here" in result.summary
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_returns_skipped_when_within_budget():
|
|
"""No compaction when tokens fit comfortably in the window."""
|
|
entries = [
|
|
{"role": "user", "content": "hi", "token_count": 5},
|
|
{"role": "assistant", "content": "hello", "token_count": 5},
|
|
]
|
|
request = CompactionRequest(
|
|
session_id="budget-test",
|
|
entries=entries,
|
|
context_window_tokens=8192,
|
|
)
|
|
result = await compact_context_new(request)
|
|
assert result.removed_count == 0
|
|
assert result.summary_source == "skipped"
|
|
assert result.kept_entries == entries
|
|
assert result.skip_reason == "within_compaction_budget"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_new_empty_entries():
|
|
"""Empty entries list returns a no-op skipped result."""
|
|
request = CompactionRequest(
|
|
session_id="empty-test",
|
|
entries=[],
|
|
context_window_tokens=8192,
|
|
)
|
|
result = await compact_context_new(request)
|
|
assert result.removed_count == 0
|
|
assert result.kept_entries == []
|
|
assert result.summary == ""
|
|
assert result.skip_reason == "no_entries"
|