| """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 |
|
|
