File size: 7,512 Bytes
108e227 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 | """Client projections must not expose model-only compaction scaffolding."""
from __future__ import annotations
from aiohttp.test_utils import TestClient, TestServer
import pytest
from agent.compaction_display import project_compaction_message_for_display
from agent.context_compressor import (
COMPRESSED_SUMMARY_METADATA_KEY,
HISTORICAL_TASK_HEADING,
SUMMARY_PREFIX,
_MERGED_PRIOR_CONTEXT_HEADER,
_MERGED_SUMMARY_DELIMITER,
_SUMMARY_END_MARKER,
)
from gateway.config import PlatformConfig
from gateway.platforms.api_server import (
APIServerAdapter,
_is_compressed_summary_message,
)
from hermes_state import SessionDB
STANDALONE_SUMMARY = (
f"{SUMMARY_PREFIX}\n\n"
f"{HISTORICAL_TASK_HEADING}\nold work\n\n"
f"{_SUMMARY_END_MARKER}"
)
MERGED_CARRIER = (
f"{_MERGED_PRIOR_CONTEXT_HEADER}\n"
"Refactor complete.\n\n"
f"{_MERGED_SUMMARY_DELIMITER}\n\n"
f"{STANDALONE_SUMMARY}"
)
REAL_USER = "test the browser controller again"
def _row(role: str, content, **extra) -> dict:
row = {"id": 1, "session_id": "s1", "role": role, "content": content}
row.update(extra)
return row
@pytest.fixture
def session_db(tmp_path):
db = SessionDB(tmp_path / "state.db")
try:
yield db
finally:
close = getattr(db, "close", None)
if callable(close):
close()
@pytest.fixture
def adapter(session_db):
adapter = APIServerAdapter(PlatformConfig(enabled=True))
adapter._session_db = session_db
return adapter
def _messages_app(adapter: APIServerAdapter):
from aiohttp import web
app = web.Application()
app.router.add_get(
"/api/sessions/{session_id}/messages",
adapter._handle_session_messages,
)
return app
class TestMessageProjection:
def test_standalone_summary_is_hidden_without_scaffolding(self):
projected = APIServerAdapter._message_response(
_row(
"user",
STANDALONE_SUMMARY,
tool_calls=[{"id": "stale"}],
reasoning="internal compression reasoning",
reasoning_content="internal compression reasoning",
reasoning_details=[{"type": "reasoning.summary", "summary": "internal"}],
codex_reasoning_items=[{"type": "reasoning", "id": "internal"}],
codex_message_items=[{"type": "message", "id": "internal"}],
)
)
assert projected["content"] == ""
assert projected["display_kind"] == "hidden"
assert "tool_calls" not in projected
assert "finish_reason" not in projected
assert "reasoning" not in projected
assert "reasoning_content" not in projected
assert "reasoning_details" not in projected
assert "codex_reasoning_items" not in projected
assert "codex_message_items" not in projected
def test_merged_carrier_preserves_only_real_prior_content(self):
projected = APIServerAdapter._message_response(
_row(
"assistant",
MERGED_CARRIER,
tool_calls=[{"id": "prior-call"}],
finish_reason="tool_calls",
)
)
assert projected["content"] == "Refactor complete."
assert "tool_calls" not in projected
assert "finish_reason" not in projected
assert "PRIOR CONTEXT" not in projected["content"]
assert "CONTEXT COMPACTION" not in projected["content"]
def test_merged_content_array_preserves_blocks_before_summary(self):
projected = APIServerAdapter._message_response(
_row(
"user",
[
{
"type": "text",
"text": f"{_MERGED_PRIOR_CONTEXT_HEADER}\n{REAL_USER}",
},
{
"type": "text",
"text": f"{_MERGED_SUMMARY_DELIMITER}\n{STANDALONE_SUMMARY}",
},
],
)
)
assert projected["content"] == [{"type": "text", "text": REAL_USER}]
def test_real_message_that_mentions_marker_text_is_untouched(self):
message = _row(
"user",
"please explain the string [CONTEXT COMPACTION] in this bug report",
tool_calls=[{"id": "real"}],
reasoning="real provider payload",
)
projected = project_compaction_message_for_display(message)
assert projected == message
assert projected is not message
def test_unrelated_hidden_message_is_not_reclassified_as_compaction(self):
message = _row("assistant", "ordinary hidden control row", display_kind="hidden")
assert _is_compressed_summary_message(message) is False
class TestSummaryRecognizer:
@pytest.mark.parametrize(
"message",
[
_row("user", STANDALONE_SUMMARY),
_row("assistant", MERGED_CARRIER),
_row("assistant", "metadata-only", **{COMPRESSED_SUMMARY_METADATA_KEY: True}),
],
)
def test_recognizes_all_compaction_carrier_shapes(self, message):
assert _is_compressed_summary_message(message) is True
def test_ignores_real_message(self):
assert _is_compressed_summary_message(_row("user", REAL_USER)) is False
class TestTurnTranscriptProjection:
def test_run_completed_strips_scaffolding_but_keeps_real_carrier_content(self):
result = {
"messages": [
{"role": "user", "content": REAL_USER},
{"role": "assistant", "content": "checking the controller"},
_row("user", STANDALONE_SUMMARY),
_row("assistant", MERGED_CARRIER),
{"role": "assistant", "content": "the controller is ready"},
],
"final_response": "the controller is ready",
}
turn = APIServerAdapter._turn_transcript_messages(
[{"role": "user", "content": REAL_USER}],
REAL_USER,
result,
)
assert [message.get("content") for message in turn] == [
"checking the controller",
"Refactor complete.",
"the controller is ready",
]
class TestMessagesEndpointProjection:
@pytest.mark.asyncio
async def test_messages_endpoint_never_serves_compaction_scaffolding(
self,
adapter,
session_db,
):
session_id = session_db.create_session("projection-session", "api_server")
session_db.replace_messages(
session_id,
[
_row("user", STANDALONE_SUMMARY),
_row("assistant", MERGED_CARRIER),
_row("user", REAL_USER),
],
)
async with TestClient(TestServer(_messages_app(adapter))) as client:
response = await client.get(f"/api/sessions/{session_id}/messages")
assert response.status == 200
payload = await response.json()
messages = payload["data"]
assert len(messages) == 3
assert messages[0]["content"] == ""
assert messages[0]["display_kind"] == "hidden"
assert messages[1]["content"] == "Refactor complete."
assert messages[2]["content"] == REAL_USER
rendered = " ".join(str(message.get("content") or "") for message in messages)
assert "PRIOR CONTEXT" not in rendered
assert "CONTEXT COMPACTION" not in rendered
|