From a5cc6e8c2c635e73df06700b7e6e89f0f5d39929 Mon Sep 17 00:00:00 2001 From: Jazzcort Date: Tue, 11 Aug 2026 13:36:05 -0400 Subject: [PATCH 1/2] Add integration tests for conversation compaction across all endpoints Cover query, A2A, responses, and streaming_query endpoints with compaction integration tests: summarization trigger, partitioning, additive summarization, marker idempotency, disabled pass-through, concurrent-request locking, and recursive fold. Supporting changes: - Add InMemoryConversationStore and _FakePage to integration conftest with dual wiring (items.list/create + conversations.items.list/create) - Document mock_conversation_store fixture in integration README --- tests/integration/README.md | 29 + tests/integration/conftest.py | 137 +- ...test_conversation_compation_integration.py | 3114 +++++++++++++++++ 3 files changed, 3277 insertions(+), 3 deletions(-) create mode 100644 tests/integration/endpoints/test_conversation_compation_integration.py diff --git a/tests/integration/README.md b/tests/integration/README.md index 6863e4869..adfe859d6 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -79,6 +79,35 @@ def test_example(mock_ogx_client: Any) -> None: mock_ogx_client.responses.create.return_value = custom_response ``` +#### `mock_conversation_store` (function-scoped) +Wires an `InMemoryConversationStore` into the mock OGX client, replacing both `conversations.items.list/create` and `items.list/create` with stateful fakes. Conversation items persist across calls within a single test, so `get_all_conversation_items` and `_write_summary_marker` work without extra patching. + +Pre-populate a conversation with `await store.create(conv_id, items=[...])` and inspect stored items via `store.store[conv_id]`. + +Requires `mock_ogx_client`. + +```python +async def test_example( + mock_ogx_client: Any, + mock_conversation_store: InMemoryConversationStore, +) -> None: + # Seed conversation items + await mock_conversation_store.create( + conversation_id="conv_abc123...", + items=[OpenAIResponseMessage(role="user", content="hello")], + ) + + # After endpoint call, verify stored items + stored = mock_conversation_store.store["conv_abc123..."] + assert len(stored) == 3 +``` + +`InMemoryConversationStore` is also importable directly for type hints or custom setups: + +```python +from tests.integration.conftest import InMemoryConversationStore +``` + ## Helper Functions Helper functions in `conftest.py` make it easier to create common test objects: diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 55e3ead36..ac0a54a2e 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -1,7 +1,10 @@ """Shared fixtures for integration tests.""" +# pylint: disable=too-many-lines + import importlib import os +from collections import defaultdict from collections.abc import AsyncIterator, Generator from pathlib import Path from typing import Any, Optional @@ -9,6 +12,7 @@ import pytest from fastapi import Request, Response from fastapi.testclient import TestClient +from ogx_client import OpenAIResponseMessage from ogx_client.models.list_models_v1_models_get200_response import ( ListModelsV1ModelsGet200Response, ) @@ -457,6 +461,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 @@ -578,6 +583,99 @@ def shutdown_integration_otel_provider(provider: TracerProvider) -> None: trace._TRACER_PROVIDER = None # pylint: disable=protected-access +# ========================================== +# 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[OpenAIResponseMessage]] = defaultdict(list) + + @property + def store(self) -> dict[str, list[OpenAIResponseMessage]]: + """Direct access to the backing store for seeding data in tests.""" + return self._store + + def _dict_to_message(self, raw: dict[str, Any]) -> OpenAIResponseMessage: + """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 OpenAIResponseMessage.model_construct( + type="message", + role=role, + content=content, + ) + + async def create( + self, + conversation_id: str, + *, + items: Any = None, + add_items_request: Any = None, + **_kwargs: Any, + ) -> Any: + """Async fake for ``client.conversations.items.create`` and ``client.items.create``.""" + if items is None and add_items_request is not None: + items = add_items_request.items + if items is None: + return + 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[OpenAIResponseMessage]) -> None: + self.data = data + self.has_more = False + self.last_id: Optional[str] = None + + 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[OpenAIResponseMessage]) -> 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 # ========================================== @@ -894,9 +992,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. @@ -912,6 +1012,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() @@ -946,6 +1052,31 @@ 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 + mock_ogx_client.items.list = store.list + mock_ogx_client.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..9fec428ef --- /dev/null +++ b/tests/integration/endpoints/test_conversation_compation_integration.py @@ -0,0 +1,3114 @@ +"""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=too-many-statements +# pylint: disable=protected-access +# pylint: disable=import-outside-toplevel + +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, MockType +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 ( + _append_previous_response_turn, + 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" +DEFAULT_SUMMARY_TEXT = "condensed earlier turns" +DEFAULT_MODEL_RESPONSE = "This is a test response about Ansible." +FOLDED_SUMMARY_TEXT = "folded summary of all earlier conversation" + + +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} + + +async def _collect_items( + store: InMemoryConversationStore, conv_id: str +) -> list[OpenAIResponseMessage]: + """Retrieve all conversation items from the in-memory store.""" + paginator = store.list(conv_id) + page = await paginator + items = list(page.data) + while page.has_next_page(): + page = await page.get_next_page() + items.extend(page.data) + return items + + +def _verify_store_content( + actual: list[OpenAIResponseMessage], expected: list[OpenAIResponseMessage] +) -> bool: + """Compare two message lists by content, role, and type fields.""" + if len(actual) != len(expected): + return False + + return all( + getattr(a, field, None) == getattr(b, field, None) + for a, b in zip(actual, expected) + for field in ("content", "role", "type") + ) + + +def _assert_marker_count( + store: InMemoryConversationStore, + conv_id: str, + expected: int, +) -> None: + """Assert the number of compaction summary markers in the store. + + Args: + store: The in-memory conversation store to inspect. + conv_id: Conversation ID to look up. + expected: Expected number of markers. + """ + markers = [ + item + for item in store.store.get(conv_id, []) + if MARKER_SENTINEL in getattr(item, "content", "") + ] + assert ( + len(markers) == expected + ), f"Expected {expected} marker(s) in store, found {len(markers)}" + + +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 + + +async def _await_lock_contention(conv_id: str, expected_waiters: int = 2) -> None: + """Wait until the per-conversation lock has the expected number of waiters.""" + from utils.conversation_compaction import _conversation_locks + + while True: + entry = _conversation_locks.get(conv_id) + if entry is not None and entry.waiters >= expected_waiters: + return + await asyncio.sleep(0) # yield to event loop + + +def _setup_query_compaction_mocks( + mocker: MockerFixture, + mock_query_agent: AsyncMockType, + items: list[Any], + summary_text: str = DEFAULT_SUMMARY_TEXT, +) -> 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. + + Returns: + The mock for ``summarize_chunk``. + """ + mock_query_agent.model.last_output_items = [ + OpenAIResponseMessage(role="assistant", content=DEFAULT_MODEL_RESPONSE) + ] + + return 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, + ), + ) + + +def _setup_fold_mocks( + mocker: MockerFixture, + cache_patch_target: str, + items: list[Any], +) -> tuple[Any, AsyncMockType, AsyncMockType]: + """Set up mocks for recursive fold tests. + + Creates a mock cache pre-loaded with two existing summaries whose combined + token count, when a third summary is added by compaction, exceeds the + compaction threshold — triggering ``_maybe_persist_fold``. + + Each existing summary has ``token_count=8`` (total 16). The new summary + from ``summarize_chunk`` adds ``token_count=6`` (total 22), which exceeds + ``context_window(200) * threshold_ratio(0.1) = 20``. + + Args: + mocker: pytest-mock fixture. + cache_patch_target: Import path of ``configured_conversation_cache``. + items: Conversation items used to set ``summarized_through_turn``. + + Returns: + Tuple of (mock_cache, mock_summarize, mock_resummarize). + """ + existing_summaries = [ + ConversationSummary( + summary_text="summary of turns 1-2", + summarized_through_turn=2, + token_count=1, + created_at="2026-08-09T00:00:00Z", + model_used=TEST_MODEL, + ), + ConversationSummary( + summary_text="summary of turns 3-4", + summarized_through_turn=4, + token_count=1, + created_at="2026-08-09T12:00:00Z", + model_used=TEST_MODEL, + ), + ] + + mock_cache = mocker.MagicMock() + mock_cache.get_summaries.return_value = existing_summaries + + mocker.patch(cache_patch_target, return_value=mock_cache) + + mock_summarize = mocker.patch( + "utils.conversation_compaction.summarize_chunk", + new_callable=mocker.AsyncMock, + return_value=ConversationSummary( + summary_text="summary of turns 5-6", + summarized_through_turn=len(items), + token_count=1000, # to trigger the persist fold + created_at="2026-08-10T00:00:00Z", + model_used=TEST_MODEL, + ), + ) + + mock_resummarize = mocker.patch( + "utils.conversation_compaction.recursively_resummarize", + new_callable=mocker.AsyncMock, + return_value=ConversationSummary( + summary_text=FOLDED_SUMMARY_TEXT, + summarized_through_turn=len(items), + token_count=1, + created_at="2026-08-10T00:00:00Z", + model_used=TEST_MODEL, + ), + ) + + return mock_cache, mock_summarize, mock_resummarize + + +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 = _setup_query_compaction_mocks(mocker, mock_query_agent, 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 2 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert not any("question two" in t for t in input_texts) + assert not any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = _setup_query_compaction_mocks(mocker, mock_query_agent, 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert any("question two" in t for t in input_texts) + assert any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _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 = _setup_query_compaction_mocks(mocker, mock_query_agent, 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + 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?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _msg("user", "Any updates?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = _setup_query_compaction_mocks(mocker, mock_query_agent, 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 0) + + 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) + # No need to check input because without triggering compaction, + # it's just the simple user input. + + # No need to check the conversation store because when we're not in + # the compaction mode, the conversation store in organized by OGX. + + @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) + # No need to check input because without triggering compaction, + # it's just the simple user input. + + # No need to check the conversation store because when we're not in + # the compaction mode, the conversation store in organized by OGX. + + @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 = _setup_query_compaction_mocks(mocker, mock_query_agent, items) + + # --- 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 2 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert not any("question two" in t for t in input_texts) + assert not any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + # --- 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() + + 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 2) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 3 + assert not any("What else can you help with?" in t for t in input_texts) + assert not any(DEFAULT_MODEL_RESPONSE in t for t in input_texts) + assert not any("question three" in t for t in input_texts) + assert not any("answer three" in t for t in input_texts) + assert sum(DEFAULT_SUMMARY_TEXT in t for t in input_texts) == 2 + assert input_texts[-1] == "Follow-up question" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 12 + expected = ( + expected + + new_items + + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "Follow-up question"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + ) + assert _verify_store_content(items_from_store, expected) + + @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={}, + ) + ) + + try: + await asyncio.wait_for(_await_lock_contention(CONV_ID_LLAMA), 10) + except TimeoutError: + pytest.fail("Task 2 never started") + + 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() + + @pytest.mark.asyncio + async def test_query_compaction_recursive_fold( + 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: + """Recursive fold triggers when cached summaries exceed the threshold. + + Verifies: + - summarize_chunk is called (new compaction triggered). + - recursively_resummarize is called (fold triggered). + - cache.replace_summaries is called to persist the fold. + - The agent receives a single folded summary in its 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 = [ + _marker("summary of turns 1-2"), + _marker("summary of turns 3-4"), + _msg("user", "question five " * 20), + _msg("assistant", "answer five " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_cache, mock_summarize, mock_resummarize = _setup_fold_mocks( + mocker, + "app.endpoints.query.configured_conversation_cache", + 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_resummarize.assert_awaited_once() + mock_cache.replace_summaries.assert_called_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(FOLDED_SUMMARY_TEXT in t for t in input_texts) == 1 + assert input_texts[-1] == "What else can you help with?" + + +# --------------------------------------------------------------------------- +# 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 + mock_agent.model.last_output_items = [ + OpenAIResponseMessage(role="assistant", content=DEFAULT_MODEL_RESPONSE) + ] + return mock_agent + + +def _setup_a2a_compaction_mocks( + mocker: MockerFixture, + items: list[Any], + summary_text: str = DEFAULT_SUMMARY_TEXT, +) -> tuple[AsyncMockType, MockType]: + """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. + + Returns: + Tuple of (mock_summarize, 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, + ), + ) + + return mock_summarize, 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_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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 2 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert not any("question two" in t for t in input_texts) + assert not any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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_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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + # # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert any("question two" in t for t in input_texts) + assert any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _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_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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + # # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + 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?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _msg("user", "Any updates?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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_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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 0) + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is False + assert isinstance(agent_params.input, str) + # No need to check input because without triggering compaction, + # it's just the simple user input. + + # No need to check the conversation store because when we're not in + # the compaction mode, the conversation store in organized by OGX. + + @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) + # No need to check input because without triggering compaction, + # it's just the simple user input. + + # No need to check the conversation store because when we're not in + # the compaction mode, the conversation store in organized by OGX. + + @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_build_agent = _setup_a2a_compaction_mocks(mocker, items) + + # --- 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + # # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 2 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert not any("question two" in t for t in input_texts) + assert not any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + # --- 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() + + request = _build_a2a_request("Follow-up question") + await handle_a2a_jsonrpc_post(request=request, auth=test_auth, mcp_headers={}) + + mock_summarize.assert_awaited_once() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 2) + + agent_params = mock_build_agent.call_args[0][1] + assert agent_params.omit_conversation is True + assert isinstance(agent_params.input, list) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 3 + assert not any("What else can you help with?" in t for t in input_texts) + assert not any(DEFAULT_MODEL_RESPONSE in t for t in input_texts) + assert not any("question three" in t for t in input_texts) + assert not any("answer three" in t for t in input_texts) + assert sum(DEFAULT_SUMMARY_TEXT in t for t in input_texts) == 2 + assert input_texts[-1] == "Follow-up question" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 12 + expected = ( + expected + + new_items + + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "Follow-up question"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + ) + assert _verify_store_content(items_from_store, expected) + + @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={}) + ) + + try: + await asyncio.wait_for(_await_lock_contention(CONV_ID_LLAMA), 10) + except TimeoutError: + pytest.fail("Task 2 never started") + + 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, +) -> MockType: + """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_result = mocker.AsyncMock(spec=ResponsesResponse) + + output_item = OpenAIResponseMessage( + role="assistant", content=DEFAULT_MODEL_RESPONSE + ) + + async def _handle_and_append( + original_request: Any, + api_params: Any, + context: Any, + ) -> Any: + """Call the real turn-append logic, then return the mock response.""" + _ = original_request + await _append_previous_response_turn(api_params, context, [output_item]) + return mock_result + + mock_handle_non_streaming_response = mocker.patch( + "app.endpoints.responses.handle_non_streaming_response", + side_effect=_handle_and_append, + ) + + return mock_handle_non_streaming_response + + +def _setup_responses_compaction_mocks( + mocker: MockerFixture, + items: list[Any], + summary_text: str = DEFAULT_SUMMARY_TEXT, +) -> AsyncMockType: + """Set up compaction-specific mocks for responses endpoint tests. + + Patches ``summarize_chunk`` 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. + + Returns: + The mock for ``summarize_chunk``. + """ + return 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, + ), + ) + + +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 = _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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is True + + # # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert len(input_texts) == 2 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert not any("question two" in t for t in input_texts) + assert not any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = _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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + api_params = mock_handle_non_streaming_response.call_args[1]["api_params"] + assert api_params.omit_conversation is True + + # # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert len(input_texts) == 4 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert any("question two" in t for t in input_texts) + assert any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _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 = _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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + 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) + + # # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert len(input_texts) == 4 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + 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?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _msg("user", "Any updates?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = _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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 0) + + 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) + # No need to check input because without triggering compaction, + # it's just the simple user input. + + # No need to check the conversation store because when we're not in + # the compaction mode, the conversation store in organized by OGX. + + @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) + # No need to check input because without triggering compaction, + # it's just the simple user input. + + # No need to check the conversation store because when we're not in + # the compaction mode, the conversation store in organized by OGX. + + @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 = _setup_responses_compaction_mocks(mocker, items) + + # --- 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + 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) + + # # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert len(input_texts) == 2 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert not any("question two" in t for t in input_texts) + assert not any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + # --- 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() + + 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 2) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in api_params.input] + assert len(input_texts) == 3 + assert not any("What else can you help with?" in t for t in input_texts) + assert not any(DEFAULT_MODEL_RESPONSE in t for t in input_texts) + assert not any("question three" in t for t in input_texts) + assert not any("answer three" in t for t in input_texts) + assert sum(DEFAULT_SUMMARY_TEXT in t for t in input_texts) == 2 + assert input_texts[-1] == "Follow-up question" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 12 + expected = ( + expected + + new_items + + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "Follow-up question"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + ) + assert _verify_store_content(items_from_store, expected) + + @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={}, + ) + ) + + try: + await asyncio.wait_for(_await_lock_contention(CONV_ID_LLAMA), 10) + except TimeoutError: + pytest.fail("Task 2 never started") + + 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_responses_compaction_recursive_fold( + 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: + """Recursive fold triggers when cached summaries exceed the threshold. + + Verifies: + - summarize_chunk is called (new compaction triggered). + - recursively_resummarize is called (fold triggered). + - cache.replace_summaries is called to persist the fold. + - The agent receives a single folded summary in its input. + """ + _enable_compaction(test_config, context_window=200, threshold_ratio=0.1) + user_id, _, _, _ = test_auth + _create_existing_conversation(patch_db_session, user_id) + + items = [ + _marker("summary of turns 1-2"), + _marker("summary of turns 3-4"), + _msg("user", "question five " * 20), + _msg("assistant", "answer five " * 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_cache, mock_summarize, mock_resummarize = _setup_fold_mocks( + mocker, + "app.endpoints.responses.configured_conversation_cache", + 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_resummarize.assert_awaited_once() + mock_cache.replace_summaries.assert_called_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(FOLDED_SUMMARY_TEXT in t for t in input_texts) == 1 + assert input_texts[-1] == "What else can you help with?" + + +# --------------------------------------------------------------------------- +# 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, + mock_streaming_query_agent: AsyncMockType, + items: list[Any], + summary_text: str = DEFAULT_SUMMARY_TEXT, +) -> 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. + + Returns: + The mock for ``summarize_chunk``. + """ + mock_streaming_query_agent.model.last_output_items = [ + OpenAIResponseMessage( + role="assistant", content="This is a test response about Ansible." + ) + ] + + return 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, + ), + ) + + +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 = _setup_streaming_compaction_mocks( + mocker, mock_streaming_query_agent, 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 2 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert not any("question two" in t for t in input_texts) + assert not any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = _setup_streaming_compaction_mocks( + mocker, mock_streaming_query_agent, 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert any("question two" in t for t in input_texts) + assert any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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, mock_streaming_query_agent, 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 = [ + _msg("user", "question one " * 20), + _msg("assistant", "answer one " * 20), + _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 = _setup_streaming_compaction_mocks( + mocker, mock_streaming_query_agent, 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 4 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + 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?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _msg("user", "Any updates?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + @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 = _setup_streaming_compaction_mocks( + mocker, mock_streaming_query_agent, 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 0) + + 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) + # No need to check input because without triggering compaction, + # it's just the simple user input. + + # No need to check the conversation store because when we're not in + # the compaction mode, the conversation store in organized by OGX. + + @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) + # No need to check input because without triggering compaction, + # it's just the simple user input. + + # No need to check the conversation store because when we're not in + # the compaction mode, the conversation store in organized by OGX. + + @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 = _setup_streaming_compaction_mocks( + mocker, mock_streaming_query_agent, items + ) + + # --- 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 1) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 2 + assert not any("question one" in t for t in input_texts) + assert not any("answer one" in t for t in input_texts) + assert not any("question two" in t for t in input_texts) + assert not any("answer two" in t for t in input_texts) + assert any(DEFAULT_SUMMARY_TEXT in t for t in input_texts) + assert input_texts[-1] == "What else can you help with?" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 7 + expected = items + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "What else can you help with?"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + assert _verify_store_content(items_from_store, expected) + + # --- 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() + + 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() + _assert_marker_count(mock_conversation_store, CONV_ID_LLAMA, 2) + + 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) + + # Check the status of the conversation we provide to the model is what we expect + input_texts = [getattr(m, "content", "") for m in agent_params.input] + assert len(input_texts) == 3 + assert not any("What else can you help with?" in t for t in input_texts) + assert not any(DEFAULT_MODEL_RESPONSE in t for t in input_texts) + assert not any("question three" in t for t in input_texts) + assert not any("answer three" in t for t in input_texts) + assert sum(DEFAULT_SUMMARY_TEXT in t for t in input_texts) == 2 + assert input_texts[-1] == "Follow-up question" + + # Check the status of the conversation store is what we expect + items_from_store = await _collect_items(mock_conversation_store, CONV_ID_LLAMA) + assert len(items_from_store) == 12 + expected = ( + expected + + new_items + + [ + _marker(DEFAULT_SUMMARY_TEXT), + _msg("user", "Follow-up question"), + _msg("assistant", DEFAULT_MODEL_RESPONSE), + ] + ) + assert _verify_store_content(items_from_store, expected) + + @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, mock_streaming_query_agent, []) + + 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?")) + + try: + await asyncio.wait_for(_await_lock_contention(CONV_ID_LLAMA), 10) + except TimeoutError: + pytest.fail("Task 2 never started") + + 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, mock_streaming_query_agent, 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, mock_streaming_query_agent, 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 + + @pytest.mark.asyncio + async def test_streaming_compaction_recursive_fold( + 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: + """Recursive fold triggers when cached summaries exceed the threshold. + + Verifies: + - summarize_chunk is called (new compaction triggered). + - recursively_resummarize is called (fold triggered). + - cache.replace_summaries is called to persist the fold. + - The agent receives a single folded summary in its input. + - A ``compaction`` SSE event is emitted. + """ + _ = 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 = [ + _marker("summary of turns 1-2"), + _marker("summary of turns 3-4"), + _msg("user", "question five " * 20), + _msg("assistant", "answer five " * 20), + ] + await mock_conversation_store.create(conversation_id=CONV_ID_LLAMA, items=items) + + mock_cache, mock_summarize, mock_resummarize = _setup_fold_mocks( + mocker, + "app.endpoints.streaming_query.configured_conversation_cache", + 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_resummarize.assert_awaited_once() + mock_cache.replace_summaries.assert_called_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 len(input_texts) == 2 + assert sum(FOLDED_SUMMARY_TEXT in t for t in input_texts) == 1 + assert input_texts[-1] == "What else can you help with?" From 08e71a0719e33047993b7b92b2d95f50c989267b Mon Sep 17 00:00:00 2001 From: Jazzcort Date: Tue, 8 Sep 2026 09:56:16 -0400 Subject: [PATCH 2/2] Fix test filename typo --- ...integration.py => test_conversation_compaction_integration.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename tests/integration/endpoints/{test_conversation_compation_integration.py => test_conversation_compaction_integration.py} (100%) diff --git a/tests/integration/endpoints/test_conversation_compation_integration.py b/tests/integration/endpoints/test_conversation_compaction_integration.py similarity index 100% rename from tests/integration/endpoints/test_conversation_compation_integration.py rename to tests/integration/endpoints/test_conversation_compaction_integration.py