"""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_suppressed_delegation_row_keeps_evidence_but_not_presentation(self, adapter, session_db, tmp_path, monkeypatch): from gateway.wake import persist_delegation_delivery monkeypatch.setenv("HERMES_HOME", str(tmp_path)) (tmp_path / "config.yaml").write_text("display: {suppress_warning_notifications: true}") sid = session_db.create_session("diagnostic-projection", "api_server") event = {"delegation_id": "early-failure", "task_failure_notice": True, "results": [{"task_index": 0, "status": "failed"}]} await persist_delegation_delivery(adapter, text="recorded worker failure", session_id=sid, evt=event) await persist_delegation_delivery(adapter, text="recorded worker failure", session_id=sid, evt=event) stored = session_db.get_messages(sid) assert len(stored) == 1 and stored[0]["content"] == "recorded worker failure" async with TestClient(TestServer(_messages_app(adapter))) as client: response = await client.get(f"/api/sessions/{sid}/messages") assert response.status == 200 payload = await response.json() assert len(payload["data"]) == 1 assert payload["data"][0]["content"] == "" assert payload["data"][0]["display_kind"] == "hidden" @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