From 1720711a0e1e7ec488e7cf0028de0a080b91866f Mon Sep 17 00:00:00 2001 From: Jazzcort Date: Tue, 11 Aug 2026 13:36:05 -0400 Subject: [PATCH] Add integration tests for conversation compaction across multiple endpoints Introduce an InMemoryConversationStore in conftest.py to provide stateful conversation item persistence during tests, and extend AsyncOgxClientHolder patches to cover a2a, responses, streaming_query, and utils.endpoints modules. Add comprehensive integration test suite covering compaction triggering, existing marker reuse, small-conversation bypass, disabled-compaction passthrough, additive multi-round summarization, concurrent request blocking, and (for streaming) SSE event ordering and start-event deduplication. --- tests/integration/conftest.py | 122 +- ...test_conversation_compation_integration.py | 2465 +++++++++++++++++ 2 files changed, 2584 insertions(+), 3 deletions(-) create mode 100644 tests/integration/endpoints/test_conversation_compation_integration.py diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index f275caac3..aa87b863b 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -1,6 +1,7 @@ """Shared fixtures for integration tests.""" import os +from collections import defaultdict from collections.abc import AsyncIterator, Generator from pathlib import Path from typing import Any, Optional @@ -10,6 +11,9 @@ from fastapi.testclient import TestClient from ogx_api.openai_responses import OpenAIResponseObject from ogx_client.types import ListModelsResponse, VersionInfo +from ogx_client.types.conversations.item_list_response import ( + OpenAIResponseMessageOutput, +) from ogx_client.types.model import Model from pydantic_ai import AgentRunResultEvent from pydantic_ai.messages import ( @@ -355,6 +359,7 @@ def mock_agent_run_stream(events: list[Any]) -> Any: """Build an async context manager that yields pydantic-ai stream events.""" async def _event_stream() -> AsyncIterator[Any]: + """Yield pre-built events one at a time.""" for event in events: yield event @@ -441,6 +446,86 @@ def set_streaming_query_agent_run( ) +# ========================================== +# In-Memory Conversation Store +# ========================================== + + +class InMemoryConversationStore: + """In-memory fake for the Llama Stack conversations.items API. + + Provides stateful ``list`` and ``create`` methods that can be wired onto a + mock ``AsyncOgxClient`` so that ``get_all_conversation_items`` and + ``_write_summary_marker`` (and ``append_turn_items_to_conversation``) work + without patching. + """ + + def __init__(self) -> None: + """Initialize an empty in-memory conversation store.""" + self._store: dict[str, list[OpenAIResponseMessageOutput]] = defaultdict(list) + + @property + def store(self) -> dict[str, list[OpenAIResponseMessageOutput]]: + """Direct access to the backing store for seeding data in tests.""" + return self._store + + def _dict_to_message(self, raw: dict[str, Any]) -> OpenAIResponseMessageOutput: + """Convert a raw item dict (as sent by production code) to a typed object.""" + content = raw.get("content", "") + if isinstance(content, list): + content = " ".join( + part.get("text", "") for part in content if isinstance(part, dict) + ) + role = raw.get("role", "user") + return OpenAIResponseMessageOutput.model_construct( + type="message", + role=role, + content=content, + ) + + async def create(self, conversation_id: str, *, items: Any, **_kwargs: Any) -> Any: + """Async fake for ``client.conversations.items.create``.""" + for raw in items: + if isinstance(raw, dict): + self._store[conversation_id].append(self._dict_to_message(raw)) + else: + self._store[conversation_id].append(raw) + + def list(self, conversation_id: str, **_kwargs: Any) -> "_FakePaginator": + """Fake for ``client.conversations.items.list`` (returns an awaitable).""" + return _FakePaginator(list(self._store.get(conversation_id, []))) + + +class _FakePage: + """Single-page result matching the ``AsyncOpenAICursorPage`` protocol.""" + + def __init__(self, data: list[OpenAIResponseMessageOutput]) -> None: + self.data = data + + def has_next_page(self) -> bool: + """Return False — the fake always returns all items in one page.""" + return False + + async def get_next_page(self) -> "_FakePage": + """Return an empty page (should never be called).""" + return _FakePage([]) + + +class _FakePaginator: # pylint: disable=too-few-public-methods + """Awaitable object matching the ``AsyncPaginator`` protocol.""" + + def __init__(self, data: list[OpenAIResponseMessageOutput]) -> None: + self._page = _FakePage(data) + + def __await__(self) -> Any: + """Allow ``await paginator`` to return the page.""" + return self._resolve().__await__() # pylint: disable=no-member + + async def _resolve(self) -> _FakePage: + """Return the single pre-built page.""" + return self._page + + # ========================================== # Fixtures # ========================================== @@ -746,9 +831,11 @@ def mock_ogx_client_fixture( defaults for integration tests. Individual tests can override specific behaviors as needed. - Patches AsyncOgxClientHolder in both app.endpoints.query and app.main - to ensure the mock is active during TestClient startup (when app.main imports - and initializes the client) and during endpoint execution. + Patches AsyncOgxClientHolder in app.endpoints.query, app.main, + app.endpoints.a2a, app.endpoints.responses, utils.endpoints, and + app.endpoints.streaming_query to ensure the mock is active during + TestClient startup (when app.main imports and initializes the + client) and during endpoint execution. Args: mocker: pytest-mock fixture used to create and patch mocks. @@ -764,6 +851,12 @@ def mock_ogx_client_fixture( mocker.patch( "app.endpoints.conversations_v1.AsyncOgxClientHolder", mock_holder_class ) + mocker.patch("app.endpoints.a2a.AsyncOgxClientHolder", mock_holder_class) + mocker.patch("app.endpoints.responses.AsyncOgxClientHolder", mock_holder_class) + mocker.patch("utils.endpoints.AsyncOgxClientHolder", mock_holder_class) + mocker.patch( + "app.endpoints.streaming_query.AsyncOgxClientHolder", mock_holder_class + ) mock_client = mocker.AsyncMock() @@ -827,6 +920,29 @@ def mock_ogx_client_fixture( yield mock_client +@pytest.fixture(name="mock_conversation_store") +def conversation_store_fixture( + mock_ogx_client: Any, +) -> InMemoryConversationStore: + """Wire an in-memory conversation store into the mock Llama Stack client. + + Replaces ``conversations.items.list`` and ``conversations.items.create`` + on the mock client with stateful fakes so that conversation items persist + across calls within a single test. Use ``await store.create(conv_id, items=...)`` + to pre-populate a conversation. + + Args: + mock_ogx_client: The mocked Llama Stack client from mock_ogx_client_fixture. + + Returns: + The InMemoryConversationStore instance backing the mock client. + """ + store = InMemoryConversationStore() + mock_ogx_client.conversations.items.list = store.list + mock_ogx_client.conversations.items.create = store.create + return store + + @pytest.fixture(name="mock_query_agent") def mock_query_agent_fixture(mocker: MockerFixture) -> Any: """Patch build_agent for /query and return the mock agent.""" diff --git a/tests/integration/endpoints/test_conversation_compation_integration.py b/tests/integration/endpoints/test_conversation_compation_integration.py new file mode 100644 index 000000000..cbf1b5907 --- /dev/null +++ b/tests/integration/endpoints/test_conversation_compation_integration.py @@ -0,0 +1,2465 @@ +"""Integration tests for conversation compaction in query, A2A, streaming, and responses.""" + +# pylint: disable=too-many-arguments +# pylint: disable=too-many-positional-arguments +# pylint: disable=too-many-lines +# pylint: disable=too-many-locals +# pylint: disable=protected-access + +import asyncio +import json +import uuid +from collections.abc import AsyncIterator +from typing import Any, cast + +import pytest +from a2a.types import ( + AgentCapabilities, + AgentCard, + AgentProvider, +) +from fastapi import Request +from ogx_api.openai_responses import OpenAIResponseMessage +from pydantic_ai import AgentRunResultEvent +from pytest_mock import AsyncMockType, MockerFixture +from sqlalchemy.orm import Session + +from app.endpoints.a2a import handle_a2a_jsonrpc_post +from app.endpoints.query import query_endpoint_handler +from app.endpoints.responses import responses_endpoint_handler +from app.endpoints.streaming_query import streaming_query_endpoint_handler +from authentication.interface import AuthTuple +from configuration import AppConfig +from models.api.requests import QueryRequest, ResponsesRequest +from models.api.responses.successful import ResponsesResponse +from models.common.responses.contexts import ResponsesContext +from models.common.responses.responses_api_params import ResponsesApiParams +from models.compaction import ConversationSummary +from models.config import CompactionConfiguration +from models.database.conversations import UserConversation +from tests.integration.conftest import InMemoryConversationStore +from utils.conversation_compaction import MARKER_SENTINEL + +EXISTING_CONV_ID = "22222222-2222-2222-2222-222222222222" +CONV_ID_LLAMA = f"conv_{EXISTING_CONV_ID}" +TEST_MODEL = "test-provider/test-model" + + +def _msg(role: str, text: str) -> OpenAIResponseMessage: + """Build a typed conversation message item.""" + return OpenAIResponseMessage(role=cast(Any, role), content=text) + + +def _marker(text: str) -> OpenAIResponseMessage: + """Build a compaction summary marker message.""" + return OpenAIResponseMessage( + role="user", + content=f"{MARKER_SENTINEL} {text}", + ) + + +def _enable_compaction( + config: AppConfig, + context_window: int = 200, + threshold_ratio: float = 0.1, + buffer_turns: int = 0, + buffer_max_ratio: float = 0.3, +) -> None: + """Override compaction and inference config to trigger compaction easily. + + Args: + config: The application configuration singleton. + context_window: Context window size for the test model. + threshold_ratio: Ratio of context window that triggers compaction. + buffer_turns: Number of recent turns to keep uncompacted. + buffer_max_ratio: Maximum ratio of context window for buffered turns. + """ + # pylint: disable=protected-access + assert config._configuration is not None + config._configuration.compaction = CompactionConfiguration( + enabled=True, + threshold_ratio=threshold_ratio, + token_floor=0, + buffer_turns=buffer_turns, + buffer_max_ratio=buffer_max_ratio, + ) + config._configuration.inference.context_windows = {TEST_MODEL: context_window} + + +def _patch_write_summary_marker( + mocker: MockerFixture, + conversation_store: "InMemoryConversationStore | None" = None, +) -> AsyncMockType: + """Patch ``_write_summary_marker`` with a fake or silent mock. + + Args: + mocker: pytest-mock fixture. + conversation_store: When provided, the patch writes a real marker + into this store; otherwise it is a silent no-op AsyncMock. + + Returns: + The mock object for ``_write_summary_marker``. + """ + if conversation_store is not None: + + async def _fake_write_marker( + client: Any, conversation_id: str, text: str + ) -> None: + _ = client + marker_item = { + "type": "message", + "role": "user", + "content": [ + {"type": "input_text", "text": f"{MARKER_SENTINEL} {text}"} + ], + } + await conversation_store.create( + conversation_id=conversation_id, items=[marker_item] + ) + + return mocker.patch( + "utils.conversation_compaction._write_summary_marker", + side_effect=_fake_write_marker, + ) + + return mocker.patch( + "utils.conversation_compaction._write_summary_marker", + new_callable=mocker.AsyncMock, + ) + + +def _patch_get_all_conversation_items(mocker: MockerFixture): + """Patch ``get_all_conversation_items`` with a slow fake for concurrency tests. + + The first call blocks until ``release`` is set; the second call signals + ``task2_entered`` and returns immediately. + + Args: + mocker: pytest-mock fixture. + + Returns: + Tuple of (entered, release, task2_entered) asyncio Events. + """ + entered = asyncio.Event() + release = asyncio.Event() + task2_entered = asyncio.Event() + + async def slow_get_items(client, conv_id): + """First call holds the lock; second call signals and returns.""" + _ = client + _ = conv_id + + if not entered.is_set(): + entered.set() + await release.wait() + else: + task2_entered.set() + return [] + + mocker.patch( + "utils.conversation_compaction.get_all_conversation_items", + side_effect=slow_get_items, + ) + + return entered, release, task2_entered + + +def _setup_query_compaction_mocks( + mocker: MockerFixture, + items: list[Any], + summary_text: str = "condensed earlier turns", + conversation_store: "InMemoryConversationStore | None" = None, +) -> tuple[AsyncMockType, AsyncMockType]: + """Set up the common compaction mocks. + + Args: + mocker: pytest-mock fixture. + items: Conversation items used to set summarized_through_turn. + summary_text: Text returned by the fake summarize_chunk. + conversation_store: When provided, _write_summary_marker writes the + marker into this store instead of being a silent no-op. This is + needed for multi-round tests that rely on markers being present + on subsequent queries. + + Returns: + Tuple of (mock_summarize, mock_write_marker). + """ + mock_summarize = mocker.patch( + "utils.conversation_compaction.summarize_chunk", + new_callable=mocker.AsyncMock, + return_value=ConversationSummary( + summary_text=summary_text, + summarized_through_turn=len(items), + token_count=6, + created_at="2026-08-10T00:00:00Z", + model_used=TEST_MODEL, + ), + ) + + mock_write_marker = _patch_write_summary_marker(mocker, conversation_store) + + return mock_summarize, mock_write_marker + + +def _create_existing_conversation( + db_session: Session, + user_id: str, +) -> None: + """Insert an existing conversation row into the test DB. + + Args: + db_session: SQLAlchemy session bound to the test database. + user_id: Owner user ID for the conversation row. + """ + conv = UserConversation( + id=EXISTING_CONV_ID, + user_id=user_id, + last_used_model="test-model", + last_used_provider="test-provider", + topic_summary="Support question", + message_count=4, + ) + db_session.add(conv) + db_session.commit() + + +class TestQueryConversationCompation: + """Tests for conversation compaction behaviour in the query endpoint.""" + + @pytest.mark.asyncio + async def test_query_compaction_triggers_summarization( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Compaction triggers summarization when tokens exceed threshold. + + Verifies: + - summarize_chunk is called for the old items + - _write_summary_marker is called to persist the marker + - The agent receives compacted params (omit_conversation=True, + explicit input with summary text and the new query) + """ + _ = mock_ogx_client + + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_query_compaction_mocks(mocker, items) + + await query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What else can you help with?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + @pytest.mark.asyncio + async def test_query_compaction_partition( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Buffer turns are preserved alongside the summary in compacted input. + + With ``buffer_turns=1``, the most recent user/assistant turn pair is + kept verbatim while older turns are summarized. + + Verifies: + - summarize_chunk and _write_summary_marker are called. + - The agent receives compacted params with the summary, the buffered + recent turn pair, and the new query (4 items total). + """ + _ = mock_ogx_client + + _enable_compaction( + test_config, + context_window=200, + threshold_ratio=0.1, + buffer_turns=1, + buffer_max_ratio=0.5, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_query_compaction_mocks(mocker, items) + + await query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What else can you help with?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert any("question two" in t for t in input_texts) + assert any("answer two" in t for t in input_texts) + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + @pytest.mark.asyncio + async def test_query_compaction_existing_marker_no_new_summarization( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Existing marker builds explicit input without new summarization. + + Verifies: + - summarize_chunk is NOT called (under threshold) + - Agent receives compacted params with summary from the marker, + recent messages, and the new query + """ + _ = mock_ogx_client + + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=1, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _marker("Summary of the earlier discussion about troubleshooting"), + _msg("user", "recent follow-up question"), + _msg("assistant", "recent follow-up answer"), + ] + + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_query_compaction_mocks(mocker, items) + + await query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="Any updates?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_not_called() + mock_write_marker.assert_not_called() + + agent_params = mock_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("Summary of the earlier discussion" in t for t in input_texts) + assert any("recent follow-up question" in t for t in input_texts) + assert any("recent follow-up answer" in t for t in input_texts) + assert input_texts[-1] == "Any updates?" + + @pytest.mark.asyncio + async def test_query_compaction_small_conversation_no_compaction( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Small conversation under threshold passes through without compaction. + + Verifies: + - No summarization or marker write + - Agent receives normal (non-compacted) params + """ + _ = mock_ogx_client + + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=4, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "hi"), + _msg("assistant", "hello"), + ] + + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_query_compaction_mocks(mocker, items) + + await query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="short question", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_not_called() + mock_write_marker.assert_not_called() + + agent_params = mock_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is False + assert isinstance(agent_params.input, str) + + @pytest.mark.asyncio + async def test_query_compaction_disabled_passes_through( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_query_agent: AsyncMockType, + patch_db_session: Session, + test_request: Request, + test_auth: AuthTuple, + ) -> None: + """Disabled compaction skips the pipeline entirely. + + Verifies: + - Agent receives unchanged, non-compacted params + """ + _ = test_config + _ = mock_ogx_client + + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + await query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What is Ansible?", conversation_id=EXISTING_CONV_ID + ), + auth=test_auth, + mcp_headers={}, + ) + + agent_params = mock_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is False + assert isinstance(agent_params.input, str) + + @pytest.mark.asyncio + async def test_query_conversation_compaction_additive_summarization( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ): + """Two successive queries produce additive summaries. + + Verifies: + - Round 1 triggers summarization and writes a marker. + - Round 2 sees the existing marker, triggers a second summarization, + and delivers both summaries in the explicit input. + """ + _ = mock_ogx_client + + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_query_compaction_mocks( + mocker, items, conversation_store=mock_conversation_store + ) + + # --- Round 1: first compaction should summarize the old items --- + await query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What else can you help with?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # --- Round 2: new turns added after the marker --- + new_items = [ + _msg("user", "question three " * 20), + _msg("assistant", "answer three " * 20), + ] + await mock_conversation_store.create( + conversation_id=CONV_ID_LLAMA, items=new_items + ) + + mock_summarize.reset_mock() + mock_write_marker.reset_mock() + + await query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="Follow-up question", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert sum("condensed earlier turns" in t for t in input_texts) == 2 + assert input_texts[-1] == "Follow-up question" + + @pytest.mark.asyncio + async def test_query_conversation_compaction_blocking_concurrent_request_with_same_id( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_query_agent: AsyncMockType, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ): + """Concurrent requests on the same conversation are serialized by the lock. + + Verifies: + - Task 2 cannot enter the compaction critical section while task 1 + holds the per-conversation lock. + - Task 2 proceeds once task 1 releases the lock. + """ + _ = mock_ogx_client + _ = mock_query_agent + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + entered, release, task2_entered = _patch_get_all_conversation_items(mocker) + + task1 = asyncio.create_task( + query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What is Ansible?", conversation_id=EXISTING_CONV_ID + ), + auth=test_auth, + mcp_headers={}, + ) + ) + await entered.wait() + + task2 = asyncio.create_task( + query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What is RHEL?", conversation_id=EXISTING_CONV_ID + ), + auth=test_auth, + mcp_headers={}, + ) + ) + await asyncio.sleep(0.05) + + assert not task2.done() + + # This proves that the second call is blocked by _conversation_locks so it does + # not even reach to slow_get_items + assert not task2_entered.is_set() + + release.set() + await asyncio.gather(task1, task2) + + assert task2_entered.is_set() + + +# --------------------------------------------------------------------------- +# A2A endpoint helpers +# --------------------------------------------------------------------------- + +_FAKE_AGENT_CARD = AgentCard( + name="Test Agent", + description="Test", + version="0.0.1", + url="http://localhost:8080/a2a", + provider=AgentProvider(organization="test", url="http://test"), + skills=[], + default_input_modes=["text/plain"], + default_output_modes=["text/plain"], + capabilities=AgentCapabilities(streaming=False), + protocol_version="0.3.0", +) + + +def _build_a2a_request(user_input: str) -> Request: + """Build a FastAPI Request with a JSON-RPC ``message/send`` body. + + Args: + user_input: The user message text to include in the A2A request. + + Returns: + A FastAPI Request object with the JSON-RPC body ready for consumption. + """ + body_dict = { + "jsonrpc": "2.0", + "id": str(uuid.uuid4()), + "method": "message/send", + "params": { + "message": { + "role": "user", + "parts": [{"type": "text", "text": user_input}], + "messageId": f"msg-{uuid.uuid4()}", + "contextId": f"ctx-{uuid.uuid4()}", + } + }, + } + body_bytes = json.dumps(body_dict).encode() + + async def receive() -> dict[str, Any]: + """Return the pre-built body as an ASGI receive event.""" + return {"type": "http.request", "body": body_bytes, "more_body": False} + + return Request( + scope={ + "type": "http", + "method": "POST", + "path": "/a2a", + "root_path": "", + "query_string": b"", + "headers": [ + (b"content-type", b"application/json"), + ], + }, + receive=receive, + ) + + +def _mock_a2a_agent(mocker: MockerFixture) -> Any: + """Build a mock pydantic-ai agent that yields a single result event. + + Args: + mocker: pytest-mock fixture. + + Returns: + A mock agent whose ``run_stream_events`` returns a single result event. + """ + mock_run_result = mocker.MagicMock() + mock_run_result.response.text = "Test A2A response" + result_event = mocker.MagicMock(spec=AgentRunResultEvent) + result_event.result = mock_run_result + + async def _event_stream() -> AsyncIterator[Any]: + """Yield a single agent run result event.""" + yield result_event + + mock_stream_ctx = mocker.AsyncMock() + mock_stream_ctx.__aenter__ = mocker.AsyncMock(return_value=_event_stream()) + mock_stream_ctx.__aexit__ = mocker.AsyncMock(return_value=False) + mock_agent = mocker.MagicMock() + mock_agent.run_stream_events.return_value = mock_stream_ctx + return mock_agent + + +def _setup_a2a_compaction_mocks( + mocker: MockerFixture, + items: list[Any], + summary_text: str = "condensed earlier turns", + conversation_store: "InMemoryConversationStore | None" = None, +) -> tuple[Any, Any, Any]: + """Set up mocks shared by A2A compaction tests. + + Patches the agent card, prepare_responses_params, and build_agent so + that ``handle_a2a_jsonrpc_post`` reaches the real + ``apply_compaction_blocking`` code path. + + Args: + mocker: pytest-mock fixture. + items: Conversation items used to set summarized_through_turn. + summary_text: Text returned by the fake summarize_chunk. + conversation_store: When provided, _write_summary_marker writes the + marker into this store instead of being a silent no-op. + + Returns: + Tuple of (mock_summarize, mock_write_marker, mock_build_agent). + """ + mocker.patch( + "app.endpoints.a2a.get_lightspeed_agent_card", + return_value=_FAKE_AGENT_CARD, + ) + + async def _fake_prepare(client, query_request, *args, **kwargs): + """Return ResponsesApiParams with the real query as input.""" + _ = client, args, kwargs + return ResponsesApiParams( + input=query_request.query, + model=TEST_MODEL, + conversation=CONV_ID_LLAMA, + store=True, + stream=True, + ) + + mocker.patch( + "app.endpoints.a2a.prepare_responses_params", + side_effect=_fake_prepare, + ) + + mock_agent = _mock_a2a_agent(mocker) + mock_build_agent = mocker.patch( + "app.endpoints.a2a.build_agent", + return_value=mock_agent, + ) + + mock_summarize = mocker.patch( + "utils.conversation_compaction.summarize_chunk", + new_callable=mocker.AsyncMock, + return_value=ConversationSummary( + summary_text=summary_text, + summarized_through_turn=len(items), + token_count=6, + created_at="2026-08-10T00:00:00Z", + model_used=TEST_MODEL, + ), + ) + + mock_write_marker = _patch_write_summary_marker(mocker, conversation_store) + + return mock_summarize, mock_write_marker, mock_build_agent + + +class TestA2AConversationCompaction: + """Tests for conversation compaction behaviour in the A2A endpoint.""" + + @pytest.mark.asyncio + async def test_a2a_compaction_triggers_summarization( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_auth: AuthTuple, + mocker: MockerFixture, + ) -> None: + """Compaction triggers summarization when tokens exceed threshold. + + Verifies: + - summarize_chunk is called for the old items + - _write_summary_marker is called to persist the marker + - The agent receives compacted params (omit_conversation=True, + explicit input with summary text and the new query) + """ + _ = mock_ogx_client + + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker, mock_build_agent = ( + _setup_a2a_compaction_mocks(mocker, items) + ) + + request = _build_a2a_request("What else can you help with?") + await handle_a2a_jsonrpc_post(request=request, auth=test_auth, mcp_headers={}) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + @pytest.mark.asyncio + async def test_a2a_compaction_partition( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_auth: AuthTuple, + mocker: MockerFixture, + ) -> None: + """Buffer turns are preserved alongside the summary in compacted input. + + With ``buffer_turns=1``, the most recent user/assistant turn pair is + kept verbatim while older turns are summarized. + + Verifies: + - summarize_chunk and _write_summary_marker are called. + - The agent receives compacted params with the summary, the buffered + recent turn pair, and the new query (4 items total). + """ + _ = mock_ogx_client + + _enable_compaction( + test_config, + context_window=200, + threshold_ratio=0.1, + buffer_turns=1, + buffer_max_ratio=0.5, + ) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker, mock_build_agent = ( + _setup_a2a_compaction_mocks(mocker, items) + ) + + request = _build_a2a_request("What else can you help with?") + await handle_a2a_jsonrpc_post(request=request, auth=test_auth, mcp_headers={}) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert any("question two" in t for t in input_texts) + assert any("answer two" in t for t in input_texts) + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + @pytest.mark.asyncio + async def test_a2a_compaction_existing_marker_no_new_summarization( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_auth: AuthTuple, + mocker: MockerFixture, + ) -> None: + """Existing marker builds explicit input without new summarization. + + Verifies: + - summarize_chunk is NOT called (under threshold) + - Agent receives compacted params with summary from the marker, + recent messages, and the new query + """ + _ = mock_ogx_client + + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=1, + ) + + items = [ + _marker("Summary of the earlier discussion about troubleshooting"), + _msg("user", "recent follow-up question"), + _msg("assistant", "recent follow-up answer"), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker, mock_build_agent = ( + _setup_a2a_compaction_mocks(mocker, items) + ) + + request = _build_a2a_request("Any updates?") + await handle_a2a_jsonrpc_post(request=request, auth=test_auth, mcp_headers={}) + + mock_summarize.assert_not_called() + mock_write_marker.assert_not_called() + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("Summary of the earlier discussion" in t for t in input_texts) + assert any("recent follow-up question" in t for t in input_texts) + assert any("recent follow-up answer" in t for t in input_texts) + assert input_texts[-1] == "Any updates?" + + @pytest.mark.asyncio + async def test_a2a_compaction_small_conversation_no_compaction( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_auth: AuthTuple, + mocker: MockerFixture, + ) -> None: + """Small conversation under threshold passes through without compaction. + + Verifies: + - No summarization or marker write + - Agent receives normal (non-compacted) params + """ + _ = mock_ogx_client + + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=4, + ) + + items = [ + _msg("user", "hi"), + _msg("assistant", "hello"), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker, mock_build_agent = ( + _setup_a2a_compaction_mocks(mocker, items) + ) + + request = _build_a2a_request("short question") + await handle_a2a_jsonrpc_post(request=request, auth=test_auth, mcp_headers={}) + + mock_summarize.assert_not_called() + mock_write_marker.assert_not_called() + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is False + assert isinstance(agent_params.input, str) + + @pytest.mark.asyncio + async def test_a2a_compaction_disabled_passes_through( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + test_auth: AuthTuple, + mocker: MockerFixture, + ) -> None: + """Disabled compaction skips the pipeline entirely. + + Verifies: + - Agent receives unchanged, non-compacted params + """ + _ = test_config + _ = mock_ogx_client + + _, _, mock_build_agent = _setup_a2a_compaction_mocks(mocker, []) + + request = _build_a2a_request("What is Ansible?") + await handle_a2a_jsonrpc_post(request=request, auth=test_auth, mcp_headers={}) + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is False + assert isinstance(agent_params.input, str) + + @pytest.mark.asyncio + async def test_a2a_compaction_additive_summarization( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_auth: AuthTuple, + mocker: MockerFixture, + ) -> None: + """Two successive A2A requests produce additive summaries. + + Verifies: + - Round 1 triggers summarization and writes a marker. + - Round 2 sees the existing marker, triggers a second summarization, + and delivers both summaries in the explicit input. + """ + _ = mock_ogx_client + + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker, mock_build_agent = ( + _setup_a2a_compaction_mocks( + mocker, items, conversation_store=mock_conversation_store + ) + ) + + # --- Round 1 --- + request = _build_a2a_request("What else can you help with?") + await handle_a2a_jsonrpc_post(request=request, auth=test_auth, mcp_headers={}) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # --- Round 2: new turns added after the marker --- + new_items = [ + _msg("user", "question three " * 20), + _msg("assistant", "answer three " * 20), + ] + await mock_conversation_store.create( + conversation_id=CONV_ID_LLAMA, items=new_items + ) + + mock_summarize.reset_mock() + mock_write_marker.reset_mock() + + request = _build_a2a_request("Follow-up question") + await handle_a2a_jsonrpc_post(request=request, auth=test_auth, mcp_headers={}) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert sum("condensed earlier turns" in t for t in input_texts) == 2 + assert input_texts[-1] == "Follow-up question" + + @pytest.mark.asyncio + async def test_a2a_compaction_blocking_concurrent_request_with_same_id( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + test_auth: AuthTuple, + mocker: MockerFixture, + ) -> None: + """Concurrent A2A requests on the same conversation are serialized by the lock. + + Verifies: + - Task 2 cannot enter the compaction critical section while task 1 + holds the per-conversation lock. + - Task 2 proceeds once task 1 releases the lock. + """ + _ = mock_ogx_client + + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + _setup_a2a_compaction_mocks(mocker, []) + + entered, release, task2_entered = _patch_get_all_conversation_items(mocker) + + request1 = _build_a2a_request("What is Ansible?") + task1 = asyncio.create_task( + handle_a2a_jsonrpc_post(request=request1, auth=test_auth, mcp_headers={}) + ) + await entered.wait() + + request2 = _build_a2a_request("What is RHEL?") + task2 = asyncio.create_task( + handle_a2a_jsonrpc_post(request=request2, auth=test_auth, mcp_headers={}) + ) + await asyncio.sleep(0.05) + + assert not task2.done() + assert not task2_entered.is_set() + + release.set() + await asyncio.gather(task1, task2) + + assert task2_entered.is_set() + + +# --------------------------------------------------------------------------- +# Responses endpoint helpers +# --------------------------------------------------------------------------- + +_RESPONSE_DUMP: dict[str, Any] = { + "id": "resp_compaction_test", + "object": "response", + "created_at": 1700000000, + "status": "completed", + "model": TEST_MODEL, + "output": [ + { + "type": "message", + "id": "msg-1", + "role": "assistant", + "status": "completed", + "content": [ + { + "type": "output_text", + "text": "Test compaction response.", + "annotations": [], + } + ], + } + ], + "usage": { + "input_tokens": 10, + "output_tokens": 5, + "total_tokens": 15, + "input_tokens_details": {"cached_tokens": 0}, + "output_tokens_details": {"reasoning_tokens": 0}, + }, +} + + +def _setup_responses_base( + mocker: MockerFixture, + mock_ogx_client: Any, +) -> Any: + """Set up the base mocks shared by all responses compaction tests. + + Configures the mock Llama Stack client (from the ``mock_ogx_client`` + fixture) with a ``responses.create`` return value that includes + ``model_dump``, and bypasses ResponsesContext validation. + + Args: + mocker: pytest-mock fixture. + mock_ogx_client: The mock Llama Stack client from the fixture. + + Returns: + The mock ``handle_non_streaming_response`` function. + """ + mock_response = mocker.MagicMock() + mock_response.id = "resp_compaction_test" + mock_output = mocker.MagicMock() + mock_output.type = "message" + mock_output.role = "assistant" + mock_output.content = "Test compaction response." + mock_output.refusal = None + mock_response.output = [mock_output] + mock_response.usage = mocker.MagicMock() + mock_response.usage.input_tokens = 10 + mock_response.usage.output_tokens = 5 + mock_response.status = "completed" + mock_response.model = TEST_MODEL + mock_response.model_dump.return_value = _RESPONSE_DUMP.copy() + mock_ogx_client.responses.create = mocker.AsyncMock(return_value=mock_response) + + original_ctx_cls = ResponsesContext + + def _skip_validation(**kwargs: Any) -> ResponsesContext: + """Bypass Pydantic validation for ResponsesContext.""" + return original_ctx_cls.model_construct(**kwargs) + + mocker.patch( + "app.endpoints.responses.ResponsesContext", side_effect=_skip_validation + ) + + mock_response = mocker.AsyncMock(spec=ResponsesResponse) + mock_handle_non_streaming_response = mocker.patch( + "app.endpoints.responses.handle_non_streaming_response", + return_value=mock_response, + ) + + return mock_handle_non_streaming_response + + +def _setup_responses_compaction_mocks( + mocker: MockerFixture, + items: list[Any], + summary_text: str = "condensed earlier turns", + conversation_store: "InMemoryConversationStore | None" = None, +) -> tuple[Any, Any]: + """Set up compaction-specific mocks for responses endpoint tests. + + Patches ``summarize_chunk`` and ``_write_summary_marker`` for compaction + integration tests. + + Args: + mocker: pytest-mock fixture. + items: Conversation items used to set summarized_through_turn. + summary_text: Text returned by the fake summarize_chunk. + conversation_store: When provided, _write_summary_marker writes the + marker into this store instead of being a silent no-op. + + Returns: + Tuple of (mock_summarize, mock_write_marker). + """ + mock_summarize = mocker.patch( + "utils.conversation_compaction.summarize_chunk", + new_callable=mocker.AsyncMock, + return_value=ConversationSummary( + summary_text=summary_text, + summarized_through_turn=len(items), + token_count=6, + created_at="2026-08-10T00:00:00Z", + model_used=TEST_MODEL, + ), + ) + + mock_write_marker = _patch_write_summary_marker(mocker, conversation_store) + + return mock_summarize, mock_write_marker + + +class TestResponsesConversationCompaction: + """Tests for conversation compaction behaviour in the responses endpoint.""" + + @pytest.mark.asyncio + async def test_responses_compaction_triggers_summarization( + self, + test_config: AppConfig, + test_auth: AuthTuple, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Compaction triggers summarization when tokens exceed threshold. + + Verifies: + - summarize_chunk is called for the old items + - _write_summary_marker is called to persist the marker + - The response completes successfully + """ + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_handle_non_streaming_response = _setup_responses_base( + mocker, mock_ogx_client + ) + + mock_summarize, mock_write_marker = _setup_responses_compaction_mocks( + mocker, items + ) + + await responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="What else can you help with?", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is True + + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + @pytest.mark.asyncio + async def test_responses_compaction_partition( + self, + test_config: AppConfig, + test_auth: AuthTuple, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Buffer turns are preserved alongside the summary in compacted input. + + With ``buffer_turns=1``, the most recent user/assistant turn pair is + kept verbatim while older turns are summarized. + + Verifies: + - summarize_chunk and _write_summary_marker are called. + - The agent receives compacted params with the summary, the buffered + recent turn pair, and the new query (4 items total). + """ + _enable_compaction( + test_config, + context_window=200, + threshold_ratio=0.1, + buffer_turns=1, + buffer_max_ratio=0.5, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_handle_non_streaming_response = _setup_responses_base( + mocker, mock_ogx_client + ) + + mock_summarize, mock_write_marker = _setup_responses_compaction_mocks( + mocker, items + ) + + await responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="What else can you help with?", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is True + + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert len(input_texts) == 4 + assert any("question two" in t for t in input_texts) + assert any("answer two" in t for t in input_texts) + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + @pytest.mark.asyncio + async def test_responses_compaction_existing_marker_no_new_summarization( + self, + test_config: AppConfig, + test_auth: AuthTuple, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Existing marker builds explicit input without new summarization. + + Verifies: + - summarize_chunk is NOT called (under threshold) + - The response completes successfully + """ + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=1, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _marker("Summary of the earlier discussion about troubleshooting"), + _msg("user", "recent follow-up question"), + _msg("assistant", "recent follow-up answer"), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_handle_non_streaming_response = _setup_responses_base( + mocker, mock_ogx_client + ) + + mock_summarize, mock_write_marker = _setup_responses_compaction_mocks( + mocker, items + ) + + await responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="Any updates?", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_not_called() + mock_write_marker.assert_not_called() + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is True + assert isinstance(api_params.input, list) + + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert any("Summary of the earlier discussion" in t for t in input_texts) + assert any("recent follow-up question" in t for t in input_texts) + assert any("recent follow-up answer" in t for t in input_texts) + assert input_texts[-1] == "Any updates?" + + @pytest.mark.asyncio + async def test_responses_compaction_small_conversation_no_compaction( + self, + test_config: AppConfig, + test_auth: AuthTuple, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Small conversation under threshold passes through without compaction. + + Verifies: + - No summarization or marker write + """ + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=4, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "hi"), + _msg("assistant", "hello"), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_handle_non_streaming_response = _setup_responses_base( + mocker, mock_ogx_client + ) + + mock_summarize, mock_write_marker = _setup_responses_compaction_mocks( + mocker, items + ) + + await responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="short question", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_not_called() + mock_write_marker.assert_not_called() + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is False + assert isinstance(api_params.input, str) + + @pytest.mark.asyncio + async def test_responses_compaction_disabled_passes_through( + self, + test_config: AppConfig, + test_auth: AuthTuple, + mock_ogx_client: AsyncMockType, + test_request: Request, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Disabled compaction skips the pipeline entirely. + + Verifies: + - The response completes without any compaction activity + """ + _ = test_config + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + mock_handle_non_streaming_response = _setup_responses_base( + mocker, mock_ogx_client + ) + + await responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="What is Ansible?", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is False + assert isinstance(api_params.input, str) + + @pytest.mark.asyncio + async def test_responses_compaction_additive_summarization( + self, + test_config: AppConfig, + test_auth: AuthTuple, + mock_ogx_client: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Two successive requests produce additive summaries. + + Verifies: + - Round 1 triggers summarization and writes a marker. + - Round 2 sees the existing marker, triggers a second summarization, + and writes a second marker. + """ + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_handle_non_streaming_response = _setup_responses_base( + mocker, mock_ogx_client + ) + + mock_summarize, mock_write_marker = _setup_responses_compaction_mocks( + mocker, items, conversation_store=mock_conversation_store + ) + + # --- Round 1 --- + await responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="What else can you help with?", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is True + assert isinstance(api_params.input, list) + + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # --- Round 2: new turns added after the marker --- + new_items = [ + _msg("user", "question three " * 20), + _msg("assistant", "answer three " * 20), + ] + await mock_conversation_store.create( + conversation_id=CONV_ID_LLAMA, items=new_items + ) + + mock_summarize.reset_mock() + mock_write_marker.reset_mock() + + await responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="Follow-up question", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is True + assert isinstance(api_params.input, list) + + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert sum("condensed earlier turns" in t for t in input_texts) == 2 + assert input_texts[-1] == "Follow-up question" + + @pytest.mark.asyncio + async def test_responses_compaction_blocking_concurrent_request_with_same_id( + self, + test_config: AppConfig, + test_auth: AuthTuple, + mock_ogx_client: AsyncMockType, + test_request: Request, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Concurrent requests on the same conversation are serialized by the lock. + + Verifies: + - Task 2 cannot enter the compaction critical section while task 1 + holds the per-conversation lock. + - Task 2 proceeds once task 1 releases the lock. + """ + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + _setup_responses_base(mocker, mock_ogx_client) + + entered, release, task2_entered = _patch_get_all_conversation_items(mocker) + + task1 = asyncio.create_task( + responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="What is Ansible?", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + ) + await entered.wait() + + task2 = asyncio.create_task( + responses_endpoint_handler( + request=test_request, + responses_request=ResponsesRequest( + input="What is RHEL?", + model=TEST_MODEL, + conversation=EXISTING_CONV_ID, + stream=False, + store=True, + generate_topic_summary=False, + ), + auth=test_auth, + mcp_headers={}, + ) + ) + await asyncio.sleep(0.05) + + assert not task2.done() + assert not task2_entered.is_set() + + release.set() + await asyncio.gather(task1, task2) + + assert task2_entered.is_set() + + +# --------------------------------------------------------------------------- +# Streaming query endpoint helpers +# --------------------------------------------------------------------------- + + +async def _collect_sse_events(response: Any) -> list[dict[str, Any]]: + """Consume a StreamingResponse and parse its SSE events into dicts. + + Args: + response: A FastAPI StreamingResponse. + + Returns: + List of parsed JSON event dicts (one per ``data:`` line). + """ + events: list[dict[str, Any]] = [] + async for chunk in response.body_iterator: + for line in chunk.strip().splitlines(): + if line.startswith("data: "): + events.append(json.loads(line[len("data: ") :])) + return events + + +def _setup_streaming_compaction_mocks( + mocker: MockerFixture, + items: list[Any], + summary_text: str = "condensed earlier turns", + conversation_store: "InMemoryConversationStore | None" = None, +) -> tuple[AsyncMockType, AsyncMockType]: + """Set up compaction mocks for streaming_query tests. + + Args: + mocker: pytest-mock fixture. + items: Conversation items used to set summarized_through_turn. + summary_text: Text returned by the fake summarize_chunk. + conversation_store: When provided, _write_summary_marker writes the + marker into this store instead of being a silent no-op. + + Returns: + Tuple of (mock_summarize, mock_write_marker). + """ + mock_summarize = mocker.patch( + "utils.conversation_compaction.summarize_chunk", + new_callable=mocker.AsyncMock, + return_value=ConversationSummary( + summary_text=summary_text, + summarized_through_turn=len(items), + token_count=6, + created_at="2026-08-10T00:00:00Z", + model_used=TEST_MODEL, + ), + ) + + mock_write_marker = _patch_write_summary_marker(mocker, conversation_store) + + return mock_summarize, mock_write_marker + + +class TestStreamingQueryConversationCompaction: + """Tests for conversation compaction behaviour in the streaming_query endpoint.""" + + @pytest.mark.asyncio + async def test_streaming_compaction_triggers_summarization( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Compaction triggers summarization when tokens exceed threshold. + + Verifies: + - summarize_chunk is called for the old items + - _write_summary_marker is called to persist the marker + - The agent receives compacted params (omit_conversation=True, + explicit input with summary text and the new query) + """ + _ = mock_ogx_client + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_streaming_compaction_mocks( + mocker, items + ) + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What else can you help with?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + _ = await _collect_sse_events(response) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_streaming_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + @pytest.mark.asyncio + async def test_streaming_compaction_partition( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Compaction triggers summarization when tokens exceed threshold. + + Verifies: + - summarize_chunk is called for the old items + - _write_summary_marker is called to persist the marker + - The agent receives compacted params (omit_conversation=True, + explicit input with summary text and the new query) + """ + _ = mock_ogx_client + _enable_compaction( + test_config, + context_window=200, + threshold_ratio=0.1, + buffer_turns=1, + buffer_max_ratio=0.5, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_streaming_compaction_mocks( + mocker, items + ) + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What else can you help with?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + _ = await _collect_sse_events(response) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + agent_params = mock_streaming_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert any("question two" in t for t in input_texts) + assert any("answer two" in t for t in input_texts) + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + @pytest.mark.asyncio + async def test_streaming_compaction_emits_start_and_compaction_sse_events( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Compaction-aware path emits ``start`` and ``compaction`` SSE events. + + Verifies: + - The first SSE event is a ``start`` event with conversation_id and + request_id. + - A ``compaction`` event with ``status: started`` is emitted before the + agent response events. + """ + _ = mock_ogx_client + _ = mock_streaming_query_agent + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + _setup_streaming_compaction_mocks(mocker, items) + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What else can you help with?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + events = await _collect_sse_events(response) + + event_types = [e.get("event") for e in events] + assert len([t for t in event_types if t == "compaction"]) == 1 + assert len([t for t in event_types if t == "start"]) == 1 + start_index, compaction_index = ( + event_types.index("start"), + event_types.index("compaction"), + ) + assert start_index < compaction_index + + start_event = next(e for e in events if e.get("event") == "start") + assert start_event["data"]["conversation_id"] == EXISTING_CONV_ID + assert "request_id" in start_event["data"] + + compaction_event = next(e for e in events if e.get("event") == "compaction") + assert compaction_event["data"]["status"] == "started" + assert compaction_event["data"]["conversation_id"] == EXISTING_CONV_ID + + @pytest.mark.asyncio + async def test_streaming_compaction_existing_marker_no_compaction_event( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Existing marker builds explicit input without emitting a compaction SSE event. + + Verifies: + - summarize_chunk is NOT called (under threshold) + - No ``compaction`` SSE event is emitted (no new summarization needed) + - Agent receives compacted params with summary from the marker + """ + _ = mock_ogx_client + _ = mock_streaming_query_agent + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=1, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _marker("Summary of the earlier discussion about troubleshooting"), + _msg("user", "recent follow-up question"), + _msg("assistant", "recent follow-up answer"), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_streaming_compaction_mocks( + mocker, items + ) + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="Any updates?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + events = await _collect_sse_events(response) + + mock_summarize.assert_not_called() + mock_write_marker.assert_not_called() + + compaction_events = [e for e in events if e.get("event") == "compaction"] + assert len(compaction_events) == 0 + + agent_params = mock_streaming_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("Summary of the earlier discussion" in t for t in input_texts) + assert input_texts[-1] == "Any updates?" + + @pytest.mark.asyncio + async def test_streaming_compaction_small_conversation_no_compaction( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Small conversation under threshold passes through without compaction. + + Verifies: + - No summarization or marker write + - No ``compaction`` SSE event is emitted + - Agent receives normal (non-compacted) params + """ + _ = mock_ogx_client + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=4, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "hi"), + _msg("assistant", "hello"), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_streaming_compaction_mocks( + mocker, items + ) + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="short question", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + events = await _collect_sse_events(response) + + mock_summarize.assert_not_called() + mock_write_marker.assert_not_called() + + compaction_events = [e for e in events if e.get("event") == "compaction"] + assert len(compaction_events) == 0 + + agent_params = mock_streaming_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is False + assert isinstance(agent_params.input, str) + + @pytest.mark.asyncio + async def test_streaming_compaction_disabled_passes_through( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + patch_db_session: Session, + test_request: Request, + test_auth: AuthTuple, + mocker: MockerFixture, + ) -> None: + """Disabled compaction skips the pipeline entirely. + + Verifies: + - Agent receives unchanged, non-compacted params + - No ``compaction`` SSE event is emitted + """ + _ = test_config + _ = mock_ogx_client + _ = mocker + + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What is Ansible?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + events = await _collect_sse_events(response) + + compaction_events = [e for e in events if e.get("event") == "compaction"] + assert len(compaction_events) == 0 + + agent_params = mock_streaming_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is False + assert isinstance(agent_params.input, str) + + @pytest.mark.asyncio + async def test_streaming_compaction_additive_summarization( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Two successive streaming queries produce additive summaries. + + Verifies: + - Round 1 triggers summarization and writes a marker. + - Round 2 sees the existing marker, triggers a second summarization, + and delivers both summaries in the explicit input. + - Both rounds emit a ``compaction`` SSE event. + """ + _ = mock_ogx_client + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_summarize, mock_write_marker = _setup_streaming_compaction_mocks( + mocker, items, conversation_store=mock_conversation_store + ) + + # --- Round 1: first compaction should summarize the old items --- + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What else can you help with?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + events = await _collect_sse_events(response) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + compaction_events = [e for e in events if e.get("event") == "compaction"] + assert len(compaction_events) == 1 + + agent_params = mock_streaming_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert any("condensed earlier turns" in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # --- Round 2: new turns added after the marker --- + new_items = [ + _msg("user", "question three " * 20), + _msg("assistant", "answer three " * 20), + ] + await mock_conversation_store.create( + conversation_id=CONV_ID_LLAMA, items=new_items + ) + + mock_summarize.reset_mock() + mock_write_marker.reset_mock() + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="Follow-up question", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + events = await _collect_sse_events(response) + + mock_summarize.assert_awaited_once() + mock_write_marker.assert_awaited_once() + + compaction_events = [e for e in events if e.get("event") == "compaction"] + assert len(compaction_events) == 1 + + agent_params = mock_streaming_query_agent.build_agent_mock.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert sum("condensed earlier turns" in t for t in input_texts) == 2 + assert input_texts[-1] == "Follow-up question" + + @pytest.mark.asyncio + async def test_streaming_compaction_blocking_concurrent_request_with_same_id( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Concurrent streaming requests on the same conversation are serialized by the lock. + + The streaming endpoint calls ``needs_compaction_path`` (unlocked) before + returning the ``StreamingResponse``, so this test patches it to True and + exercises the per-conversation lock inside ``apply_compaction`` only. + + Verifies: + - Task 2 cannot enter the compaction critical section while task 1 + holds the per-conversation lock. + - Task 2 proceeds once task 1 releases the lock. + """ + _ = mock_ogx_client + _ = mock_streaming_query_agent + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + _setup_streaming_compaction_mocks(mocker, []) + + mocker.patch( + "app.endpoints.streaming_query.needs_compaction_path", + new_callable=mocker.AsyncMock, + return_value=True, + ) + + entered, release, task2_entered = _patch_get_all_conversation_items(mocker) + + async def _run_and_drain(query: str) -> None: + """Run streaming_query_endpoint_handler and drain the response.""" + resp = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query=query, conversation_id=EXISTING_CONV_ID + ), + auth=test_auth, + mcp_headers={}, + ) + async for _ in resp.body_iterator: + pass + + task1 = asyncio.create_task(_run_and_drain("What is Ansible?")) + await entered.wait() + + task2 = asyncio.create_task(_run_and_drain("What is RHEL?")) + await asyncio.sleep(0.05) + + assert not task2.done() + assert not task2_entered.is_set() + + release.set() + await asyncio.gather(task1, task2) + + assert task2_entered.is_set() + + @pytest.mark.asyncio + async def test_streaming_compaction_sse_event_ordering( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """SSE events follow the expected order: start, compaction, agent events. + + Verifies: + - The ``start`` event comes first. + - The ``compaction`` event comes second, before any agent content events. + - Agent content events (token, turn_complete) follow the compaction event. + """ + _ = mock_ogx_client + _ = mock_streaming_query_agent + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _msg("user", "question two " * 20), + _msg("assistant", "answer two " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + _setup_streaming_compaction_mocks(mocker, items) + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="What else can you help with?", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + events = await _collect_sse_events(response) + event_types = [e.get("event") for e in events] + + assert event_types[0] == "start" + assert event_types[1] == "compaction" + + remaining_types = set(event_types[2:]) + assert remaining_types.issubset({"token", "turn_complete", "end", "error"}) + + @pytest.mark.asyncio + async def test_streaming_no_compaction_no_start_event_duplication( + self, + test_config: AppConfig, + mock_ogx_client: AsyncMockType, + mock_streaming_query_agent: AsyncMockType, + mock_conversation_store: InMemoryConversationStore, + test_request: Request, + test_auth: AuthTuple, + patch_db_session: Session, + mocker: MockerFixture, + ) -> None: + """Non-compaction path emits exactly one ``start`` event and no ``compaction`` event. + + Verifies: + - Exactly one ``start`` SSE event is emitted. + - Zero ``compaction`` SSE events are emitted. + """ + _ = mock_ogx_client + _ = mock_streaming_query_agent + _enable_compaction( + test_config, + context_window=1_000_000, + threshold_ratio=0.5, + buffer_turns=4, + ) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _msg("user", "hi"), + _msg("assistant", "hello"), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + _setup_streaming_compaction_mocks(mocker, items) + + response = await streaming_query_endpoint_handler( + request=test_request, + query_request=QueryRequest( + query="short question", + conversation_id=EXISTING_CONV_ID, + ), + auth=test_auth, + mcp_headers={}, + ) + + events = await _collect_sse_events(response) + + start_events = [e for e in events if e.get("event") == "start"] + assert len(start_events) == 1 + + compaction_events = [e for e in events if e.get("event") == "compaction"] + assert len(compaction_events) == 0