From c1fddc067b8b6f477fad574570a3eb1a3ee2a653 Mon Sep 17 00:00:00 2001 From: Maxim Svistunov Date: Fri, 14 Aug 2026 07:48:06 +0200 Subject: [PATCH 1/2] LCORE-1573: surface compaction outcome as context_status in query responses Add a context_status field ("full" when no compaction occurred, "summarized" when older turns were replaced by a compaction summary) to the two response surfaces clients actually receive on the wire: - QueryResponse (src/models/api/responses/successful/query.py) for the non-streaming /v1/query endpoint. - EndEventData (src/models/common/agents/stream_payloads.py), the SSE "end" event payload, for the streaming /v1/streaming_query endpoint. StreamingQueryResponse is a documentation-only class with an empty body, so the field is deliberately NOT added there; only its SSE example string is updated to show context_status in the end event. The value maps directly from CompactionResult.compacted (set by the LCORE-1572 compaction integration): a new context_status property on CompactionResult performs the mapping in one place. The non-streaming endpoint reads it when building QueryResponse; the streaming compaction-aware path captures it from the yielded CompactionResult and threads it through generate_agent_response (new context_status parameter, defaulting to "full" for the non-compaction path) into EndStreamPayload.create. The shared ContextStatus Literal["full", "summarized"] type alias lives in models/common/turn_summary.py (imported by both response surfaces already) and is exported from models.common. In the regenerated OpenAPI schema it becomes a named enum component referenced by QueryResponse.context_status; the streaming endpoint's SSE example now shows context_status in the end event (the streaming response is documented via an inline example only, so EndEventData itself does not appear as a component schema). /v1/responses intentionally does not get the field (it stays OpenAI-shaped and compacts silently by design, R12), and the A2A executor is out of scope for the UI-indicator use case. Unit tests cover the CompactionResult mapping, the QueryResponse field (default, explicit value, rejection of unknown values), the end event payload contents for both statuses, and the full/summarized threading through both endpoint pipelines. --- docs/devel_doc/openapi.json | 22 ++++- src/app/endpoints/query.py | 1 + src/app/endpoints/streaming_query.py | 4 + src/models/api/responses/successful/query.py | 14 ++- src/models/common/__init__.py | 2 + src/models/common/agents/stream_payloads.py | 13 ++- src/models/common/turn_summary.py | 9 +- src/utils/agents/streaming.py | 7 +- src/utils/conversation_compaction.py | 6 ++ tests/unit/app/endpoints/test_query.py | 85 ++++++++++++++++- .../app/endpoints/test_streaming_query.py | 91 +++++++++++++++++++ .../models/responses/test_query_response.py | 24 ++++- tests/unit/utils/agents/test_streaming.py | 64 +++++++++++++ .../utils/test_conversation_compaction.py | 7 ++ 14 files changed, 341 insertions(+), 8 deletions(-) diff --git a/docs/devel_doc/openapi.json b/docs/devel_doc/openapi.json index e57645739..6bb7058f2 100644 --- a/docs/devel_doc/openapi.json +++ b/docs/devel_doc/openapi.json @@ -5860,6 +5860,7 @@ "ClusterQuotaLimiter": 998911, "UserQuotaLimiter": 998911 }, + "context_status": "full", "conversation_id": "123e4567-e89b-12d3-a456-426614174000", "input_tokens": 123, "output_tokens": 456, @@ -6265,7 +6266,7 @@ "schema": { "type": "string" }, - "example": "data: {\"event\": \"start\", \"data\": {\"conversation_id\": \"123e4567-e89b-12d3-a456-426614174000\", \"request_id\": \"123e4567-e89b-12d3-a456-426614174001\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 0, \"token\": \"No Violation\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 1, \"token\": \"\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 2, \"token\": \"Hello\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 3, \"token\": \"!\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 4, \"token\": \" How\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 5, \"token\": \" can\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 6, \"token\": \" I\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 7, \"token\": \" assist\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 8, \"token\": \" you\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 9, \"token\": \" today\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 10, \"token\": \"?\"}}\n\ndata: {\"event\": \"turn_complete\", \"data\": {\"token\": \"Hello! How can I assist you today?\"}}\n\ndata: {\"event\": \"end\", \"data\": {\"referenced_documents\": [], \"truncated\": null, \"input_tokens\": 11, \"output_tokens\": 19}, \"available_quotas\": {}}\n\n" + "example": "data: {\"event\": \"start\", \"data\": {\"conversation_id\": \"123e4567-e89b-12d3-a456-426614174000\", \"request_id\": \"123e4567-e89b-12d3-a456-426614174001\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 0, \"token\": \"No Violation\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 1, \"token\": \"\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 2, \"token\": \"Hello\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 3, \"token\": \"!\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 4, \"token\": \" How\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 5, \"token\": \" can\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 6, \"token\": \" I\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 7, \"token\": \" assist\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 8, \"token\": \" you\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 9, \"token\": \" today\"}}\n\ndata: {\"event\": \"token\", \"data\": {\"id\": 10, \"token\": \"?\"}}\n\ndata: {\"event\": \"turn_complete\", \"data\": {\"token\": \"Hello! How can I assist you today?\"}}\n\ndata: {\"event\": \"end\", \"data\": {\"referenced_documents\": [], \"truncated\": null, \"context_status\": \"full\", \"input_tokens\": 11, \"output_tokens\": 19}, \"available_quotas\": {}}\n\n" } } }, @@ -13503,6 +13504,13 @@ } ] }, + "ContextStatus": { + "type": "string", + "enum": [ + "full", + "summarized" + ] + }, "ConversationData": { "properties": { "conversation_id": { @@ -18884,6 +18892,15 @@ true ] }, + "context_status": { + "$ref": "#/components/schemas/ContextStatus", + "description": "Context status: \"full\" (no compaction) or \"summarized\" (older turns replaced by a summary)", + "default": "full", + "examples": [ + "full", + "summarized" + ] + }, "input_tokens": { "type": "integer", "title": "Input Tokens", @@ -18942,13 +18959,14 @@ "response" ], "title": "QueryResponse", - "description": "Model representing LLM response to a query.\n\nAttributes:\n conversation_id: The optional conversation ID (UUID).\n response: The response.\n rag_chunks: Deprecated. List of RAG chunks used to generate the response.\n This information is now available in tool_results under file_search_call type.\n referenced_documents: The URLs and titles for the documents used to generate the response.\n tool_calls: List of tool calls made during response generation.\n tool_results: List of tool results.\n truncated: Whether conversation history was truncated.\n input_tokens: Number of tokens sent to LLM.\n output_tokens: Number of tokens received from LLM.\n available_quotas: Quota available as measured by all configured quota limiters.", + "description": "Model representing LLM response to a query.\n\nAttributes:\n conversation_id: The optional conversation ID (UUID).\n response: The response.\n rag_chunks: Deprecated. List of RAG chunks used to generate the response.\n This information is now available in tool_results under file_search_call type.\n referenced_documents: The URLs and titles for the documents used to generate the response.\n tool_calls: List of tool calls made during response generation.\n tool_results: List of tool results.\n truncated: Whether conversation history was truncated.\n context_status: Whether the conversation context was sent in full\n (\"full\") or older turns were replaced by a summary (\"summarized\").\n input_tokens: Number of tokens sent to LLM.\n output_tokens: Number of tokens received from LLM.\n available_quotas: Quota available as measured by all configured quota limiters.", "examples": [ { "available_quotas": { "ClusterQuotaLimiter": 998911, "UserQuotaLimiter": 998911 }, + "context_status": "full", "conversation_id": "123e4567-e89b-12d3-a456-426614174000", "input_tokens": 123, "output_tokens": 456, diff --git a/src/app/endpoints/query.py b/src/app/endpoints/query.py index 35f62b2f0..dd214efc2 100644 --- a/src/app/endpoints/query.py +++ b/src/app/endpoints/query.py @@ -356,6 +356,7 @@ async def _handle_query_with_tracing( rag_chunks=turn_summary.rag_chunks, referenced_documents=turn_summary.referenced_documents, truncated=False, + context_status=compaction.context_status, input_tokens=turn_summary.token_usage.input_tokens, output_tokens=turn_summary.token_usage.output_tokens, available_quotas=available_quotas, diff --git a/src/app/endpoints/streaming_query.py b/src/app/endpoints/streaming_query.py index 02a6f5479..0f9faed6f 100644 --- a/src/app/endpoints/streaming_query.py +++ b/src/app/endpoints/streaming_query.py @@ -48,6 +48,7 @@ from models.common.responses.contexts import ResponseGeneratorContext from models.common.responses.responses_api_params import ResponsesApiParams from models.common.responses.types import ResponseInput +from models.common.turn_summary import ContextStatus from models.config import Action from utils.agents.streaming import ( generate_agent_response, @@ -431,6 +432,7 @@ async def generate_response_with_compaction( ) compacted_original_input: Optional[ResponseInput] = None + context_status: ContextStatus = "full" try: async for item in apply_compaction( context.client, @@ -447,6 +449,7 @@ async def generate_response_with_compaction( elif isinstance(item, CompactionResult): responses_params = item.params compacted_original_input = item.original_input + context_status = item.context_status generator, turn_summary = await retrieve_agent_response_generator( responses_params=responses_params, @@ -495,6 +498,7 @@ async def generate_response_with_compaction( emit_start=False, original_input=compacted_original_input, root_span=root_span, + context_status=context_status, ): yield event finally: diff --git a/src/models/api/responses/successful/query.py b/src/models/api/responses/successful/query.py index c59bac766..2bd718b74 100644 --- a/src/models/api/responses/successful/query.py +++ b/src/models/api/responses/successful/query.py @@ -9,6 +9,7 @@ from models.api.responses.constants import SUCCESSFUL_RESPONSE_DESCRIPTION from models.api.responses.successful.bases import AbstractSuccessfulResponse from models.common.turn_summary import ( + ContextStatus, RAGChunk, ReferencedDocument, ToolCallSummary, @@ -28,6 +29,8 @@ class QueryResponse(AbstractSuccessfulResponse): tool_calls: List of tool calls made during response generation. tool_results: List of tool results. truncated: Whether conversation history was truncated. + context_status: Whether the conversation context was sent in full + ("full") or older turns were replaced by a summary ("summarized"). input_tokens: Number of tokens sent to LLM. output_tokens: Number of tokens received from LLM. available_quotas: Quota available as measured by all configured quota limiters. @@ -71,6 +74,13 @@ class QueryResponse(AbstractSuccessfulResponse): examples=[False, True], ) + context_status: ContextStatus = Field( + "full", + description='Context status: "full" (no compaction) or ' + '"summarized" (older turns replaced by a summary)', + examples=["full", "summarized"], + ) + input_tokens: int = Field( 0, description="Number of tokens sent to LLM", @@ -113,6 +123,7 @@ class QueryResponse(AbstractSuccessfulResponse): }, ], "truncated": False, + "context_status": "full", "input_tokens": 123, "output_tokens": 456, "available_quotas": { @@ -198,7 +209,8 @@ def openapi_response(cls) -> dict[str, Any]: '"token": "Hello! How can I assist you today?"}}\n\n' 'data: {"event": "end", "data": {' '"referenced_documents": [], ' - '"truncated": null, "input_tokens": 11, "output_tokens": 19}, ' + '"truncated": null, "context_status": "full", ' + '"input_tokens": 11, "output_tokens": 19}, ' '"available_quotas": {}}\n\n' ), ] diff --git a/src/models/common/__init__.py b/src/models/common/__init__.py index aa3a40783..deb19bff5 100644 --- a/src/models/common/__init__.py +++ b/src/models/common/__init__.py @@ -28,6 +28,7 @@ ) from models.common.transcripts import Transcript, TranscriptMetadata from models.common.turn_summary import ( + ContextStatus, MCPListToolsSummary, RAGChunk, RAGContext, @@ -44,6 +45,7 @@ "CatalogShield", "CatalogTool", "CatalogToolParameter", + "ContextStatus", "ConversationData", "ConversationDetails", "ConversationTurn", diff --git a/src/models/common/agents/stream_payloads.py b/src/models/common/agents/stream_payloads.py index f57eb2ba9..aab799d7b 100644 --- a/src/models/common/agents/stream_payloads.py +++ b/src/models/common/agents/stream_payloads.py @@ -6,7 +6,12 @@ from pydantic import BaseModel, ConfigDict, Field from models.api.responses.error import AbstractErrorResponse -from models.common import ReferencedDocument, ToolCallSummary, ToolResultSummary +from models.common import ( + ContextStatus, + ReferencedDocument, + ToolCallSummary, + ToolResultSummary, +) class StreamPayloadBase(BaseModel): @@ -49,6 +54,7 @@ class EndEventData(BaseModel): referenced_documents: list[ReferencedDocument] truncated: Optional[bool] + context_status: ContextStatus = "full" input_tokens: int output_tokens: int @@ -149,6 +155,7 @@ def create( cls, *, referenced_documents: list[ReferencedDocument], + context_status: ContextStatus, input_tokens: int, output_tokens: int, available_quotas: dict[str, int], @@ -157,6 +164,9 @@ def create( Args: referenced_documents: Documents referenced during the turn. + context_status: Whether the conversation context was sent in full + ("full") or older turns were replaced by a summary + ("summarized"). input_tokens: Input token count for the turn. output_tokens: Output token count for the turn. available_quotas: Remaining quota limits by quota name. @@ -168,6 +178,7 @@ def create( data=EndEventData( referenced_documents=referenced_documents, truncated=None, + context_status=context_status, input_tokens=input_tokens, output_tokens=output_tokens, ), diff --git a/src/models/common/turn_summary.py b/src/models/common/turn_summary.py index 37a4a8f47..092ec286c 100644 --- a/src/models/common/turn_summary.py +++ b/src/models/common/turn_summary.py @@ -3,13 +3,20 @@ Used on query and streaming paths. """ -from typing import Any, Optional +from typing import Any, Literal, Optional from ogx_api import OpenAIResponseOutput from pydantic import AnyUrl, BaseModel, Field from utils.token_counter import TokenCounter +type ContextStatus = Literal["full", "summarized"] +"""How the conversation context was assembled for a turn. + +``"full"`` means the full history was used; ``"summarized"`` means older +turns were replaced by a compaction summary (LCORE-1573). +""" + class RAGChunk(BaseModel): """Model representing a RAG chunk used in the response.""" diff --git a/src/utils/agents/streaming.py b/src/utils/agents/streaming.py index 90c33699c..8bbc19cce 100644 --- a/src/utils/agents/streaming.py +++ b/src/utils/agents/streaming.py @@ -46,7 +46,7 @@ from models.common.responses import ResponseInput from models.common.responses.contexts import ResponseGeneratorContext from models.common.responses.responses_api_params import ResponsesApiParams -from models.common.turn_summary import TurnSummary +from models.common.turn_summary import ContextStatus, TurnSummary from utils.agents.error_handler import map_agent_inference_error from utils.agents.query import ( AgentFinishReason, @@ -170,6 +170,7 @@ async def generate_agent_response( # pylint: disable=too-many-statements emit_start: bool = True, original_input: Optional[ResponseInput] = None, root_span: Optional[trace.Span] = None, + context_status: ContextStatus = "full", ) -> AsyncIterator[str]: """Wrap an agent SSE generator with cleanup logic. @@ -189,6 +190,9 @@ async def generate_agent_response( # pylint: disable=too-many-statements explicit-input rewrite. Used to persist the completed turn with its structured input (preserving attachments); ``None`` otherwise. root_span: OpenTelemetry root span for this request. + context_status: Whether the conversation context was sent in full + ("full") or older turns were replaced by a summary ("summarized"). + Reported to the client in the SSE end event. Yields: SSE-formatted strings from the wrapped generator. @@ -298,6 +302,7 @@ async def generate_agent_response( # pylint: disable=too-many-statements ) end_payload = EndStreamPayload.create( referenced_documents=turn_summary.referenced_documents, + context_status=context_status, input_tokens=turn_summary.token_usage.input_tokens, output_tokens=turn_summary.token_usage.output_tokens, available_quotas=available_quotas, diff --git a/src/utils/conversation_compaction.py b/src/utils/conversation_compaction.py index e45cefda4..ab8f4f41f 100644 --- a/src/utils/conversation_compaction.py +++ b/src/utils/conversation_compaction.py @@ -57,6 +57,7 @@ from log import get_logger from models.common.responses.responses_api_params import ResponsesApiParams from models.common.responses.types import ResponseInput +from models.common.turn_summary import ContextStatus from models.compaction import ConversationSummary from models.config import CompactionConfiguration, InferenceConfiguration from utils.compaction import ( @@ -181,6 +182,11 @@ class CompactionResult: compacted: bool original_input: Optional[ResponseInput] = None + @property + def context_status(self) -> ContextStatus: + """The API ``context_status`` value for this result (LCORE-1573).""" + return "summarized" if self.compacted else "full" + def is_marker_item(item: Any) -> bool: """Return True when *item* is a compaction summary marker message.""" diff --git a/tests/unit/app/endpoints/test_query.py b/tests/unit/app/endpoints/test_query.py index 902ad1701..edc04f8f7 100644 --- a/tests/unit/app/endpoints/test_query.py +++ b/tests/unit/app/endpoints/test_query.py @@ -1,7 +1,7 @@ # pylint: disable=too-many-locals """Unit tests for the /query (v2) REST API endpoint using Responses API.""" -from typing import Any +from typing import Any, cast import pytest from fastapi import Request @@ -22,6 +22,7 @@ TurnSummary, ) from models.database.conversations import UserConversation +from utils.conversation_compaction import CompactionResult # User ID must be proper UUID MOCK_AUTH = ( @@ -175,6 +176,88 @@ async def mock_retrieve_agent_response( assert isinstance(response, QueryResponse) assert response.conversation_id == "123" assert response.response == "Kubernetes is a container orchestration platform" + assert response.context_status == "full" + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("compacted", "expected_status"), + [(False, "full"), (True, "summarized")], + ) + async def test_query_reports_context_status( + self, + dummy_request: Request, + setup_configuration: AppConfig, + mocker: MockerFixture, + compacted: bool, + expected_status: str, + ) -> None: + """Test that the compaction outcome is surfaced as context_status.""" + query_request = QueryRequest( + query="What is Kubernetes?" + ) # pyright: ignore[reportCallIssue] + + mocker.patch("app.endpoints.query.configuration", setup_configuration) + mocker.patch("app.endpoints.query.check_configuration_loaded") + mocker.patch("app.endpoints.query.check_tokens_available") + mocker.patch("app.endpoints.query.validate_model_provider_override") + + mock_client = mocker.AsyncMock(spec=AsyncOgxClient) + mock_client_holder = mocker.Mock() + mock_client_holder.get_client.return_value = mock_client + mocker.patch( + "app.endpoints.query.AsyncOgxClientHolder", + return_value=mock_client_holder, + ) + mocker.patch( + "app.endpoints.query.maybe_get_topic_summary", + new=mocker.AsyncMock(return_value=None), + ) + mocker.patch( + "app.endpoints.query.run_shield_moderation", + new=mocker.AsyncMock(return_value=ShieldModerationPassed()), + ) + + mock_responses_params = mocker.Mock(spec=ResponsesApiParams) + mock_responses_params.model = "provider1/model1" + mock_responses_params.conversation = "conv_123" + mock_responses_params.tools = None + mocker.patch( + "app.endpoints.query.prepare_responses_params", + new=mocker.AsyncMock(return_value=mock_responses_params), + ) + + compaction_result = CompactionResult( + cast(ResponsesApiParams, mock_responses_params), + compacted=compacted, + ) + mocker.patch( + "app.endpoints.query.apply_compaction_blocking", + new=mocker.AsyncMock(return_value=compaction_result), + ) + + mock_turn_summary = TurnSummary() + mock_turn_summary.llm_response = "An answer" + mocker.patch( + "app.endpoints.query.retrieve_agent_response", + new=mocker.AsyncMock(return_value=mock_turn_summary), + ) + + mocker.patch( + "app.endpoints.query.normalize_conversation_id", return_value="123" + ) + mocker.patch("app.endpoints.query.store_query_results") + mocker.patch("app.endpoints.query.consume_query_tokens") + mocker.patch("app.endpoints.query.get_available_quotas", return_value={}) + + response = await query_endpoint_handler( + request=dummy_request, + query_request=query_request, + auth=MOCK_AUTH, + mcp_headers={}, + ) + + assert isinstance(response, QueryResponse) + assert response.context_status == expected_status @pytest.mark.asyncio async def test_query_merges_inline_and_tool_rag_chunks_and_documents( diff --git a/tests/unit/app/endpoints/test_streaming_query.py b/tests/unit/app/endpoints/test_streaming_query.py index b19d40884..990772194 100644 --- a/tests/unit/app/endpoints/test_streaming_query.py +++ b/tests/unit/app/endpoints/test_streaming_query.py @@ -14,6 +14,7 @@ from pytest_mock import MockerFixture from app.endpoints.streaming_query import ( + generate_response_with_compaction, streaming_query_endpoint_handler, ) from configuration import AppConfig @@ -30,6 +31,7 @@ TurnSummary, ) from models.config import Action +from utils.conversation_compaction import CompactionResult from utils.otel_tracing import SpanAttributes, SpanEvents INTERRUPTED_INDICATOR = f"\n\n*{INTERRUPTED_RESPONSE_MESSAGE}*" @@ -907,3 +909,92 @@ async def mock_generate_with_child( assert child.parent is not None # pyright narrowing assert root_spans[0].context is not None # pyright narrowing assert child.parent.span_id == root_spans[0].context.span_id + + +class TestGenerateResponseWithCompaction: # pylint: disable=too-few-public-methods + """Tests for the compaction-aware SSE generator.""" + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("compacted", "expected_status"), + [(False, "full"), (True, "summarized")], + ) + async def test_threads_context_status_to_agent_response( + self, + mocker: MockerFixture, + compacted: bool, + expected_status: str, + ) -> None: + """Test the CompactionResult outcome reaches generate_agent_response.""" + responses_params = ResponsesApiParams.model_validate( + { + "model": "provider1/model1", + "input": "What is OpenShift?", + "conversation": "conv_123", + "stream": True, + "store": True, + } + ) + + context = mocker.Mock() + context.conversation_id = "conv_123" + context.request_id = "req_123" + context.user_id = "user_123" + context.skip_userid_check = False + context.client = mocker.AsyncMock() + context.moderation_result = ShieldModerationPassed() + context.inline_rag_context = RAGContext() + context.query_request = QueryRequest( + query="What is OpenShift?" + ) # pyright: ignore[reportCallIssue] + + compaction_result = CompactionResult(responses_params, compacted=compacted) + + async def fake_apply_compaction( + *_args: Any, **_kwargs: Any + ) -> AsyncIterator[CompactionResult]: + yield compaction_result + + mocker.patch( + "app.endpoints.streaming_query.apply_compaction", + new=fake_apply_compaction, + ) + mocker.patch( + "app.endpoints.streaming_query.configured_conversation_cache", + return_value=None, + ) + mock_config = mocker.Mock() + mocker.patch("app.endpoints.streaming_query.configuration", mock_config) + + async def inner_generator() -> AsyncIterator[str]: + yield "data: test\n\n" + + mocker.patch( + "app.endpoints.streaming_query.retrieve_agent_response_generator", + new=mocker.AsyncMock(return_value=(inner_generator(), TurnSummary())), + ) + + captured_kwargs: dict[str, Any] = {} + + async def fake_generate_agent_response( + *_args: Any, **kwargs: Any + ) -> AsyncIterator[str]: + captured_kwargs.update(kwargs) + yield "data: end\n\n" + + mocker.patch( + "app.endpoints.streaming_query.generate_agent_response", + new=fake_generate_agent_response, + ) + + events = [ + event + async for event in generate_response_with_compaction( + context=context, + responses_params=responses_params, + endpoint_path="/v1/streaming_query", + ) + ] + + assert events # the start event plus the delegated events + assert captured_kwargs["context_status"] == expected_status diff --git a/tests/unit/models/responses/test_query_response.py b/tests/unit/models/responses/test_query_response.py index ce547ec1a..9ab9f7ac1 100644 --- a/tests/unit/models/responses/test_query_response.py +++ b/tests/unit/models/responses/test_query_response.py @@ -1,6 +1,7 @@ """Unit tests for QueryResponse model.""" -from pydantic import AnyUrl +import pytest +from pydantic import AnyUrl, ValidationError from models.api.responses.successful import QueryResponse from models.common.turn_summary import ( @@ -33,6 +34,27 @@ def test_optional_conversation_id(self) -> None: assert qr.conversation_id is None assert qr.response == "LLM answer" + def test_context_status_defaults_to_full(self) -> None: + """Test that context_status defaults to "full" when not provided.""" + qr = QueryResponse(response="LLM answer") # type: ignore[call-arg] + assert qr.context_status == "full" + + def test_context_status_summarized(self) -> None: + """Test that context_status accepts the "summarized" value.""" + qr = QueryResponse( # type: ignore[call-arg] + response="LLM answer", + context_status="summarized", + ) + assert qr.context_status == "summarized" + + def test_context_status_rejects_unknown_value(self) -> None: + """Test that context_status rejects values outside full/summarized.""" + with pytest.raises(ValidationError): + QueryResponse( # type: ignore[call-arg] + response="LLM answer", + context_status="partial", # type: ignore[arg-type] + ) + def test_complete_query_response_with_all_fields(self) -> None: """Test QueryResponse with all fields including tool calls, and tool results.""" tool_calls = [ diff --git a/tests/unit/utils/agents/test_streaming.py b/tests/unit/utils/agents/test_streaming.py index 16ca0a899..ff7f58ffc 100644 --- a/tests/unit/utils/agents/test_streaming.py +++ b/tests/unit/utils/agents/test_streaming.py @@ -678,6 +678,70 @@ async def inner() -> AsyncIterator[str]: consume_mock.assert_called_once() store_mock.assert_called_once() + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("generate_kwargs", "expected_status"), + [ + ({}, "full"), + ({"context_status": "summarized"}, "summarized"), + ], + ) + async def test_end_event_reports_context_status( + self, + mocker: MockerFixture, + make_generator_context: Callable[..., ResponseGeneratorContext], + responses_params: ResponsesApiParams, + generate_kwargs: dict[str, Any], + expected_status: str, + ) -> None: + """Test the end event carries context_status ("full" by default).""" + context = make_generator_context() + turn_summary = TurnSummary() + turn_summary.token_usage = TokenCounter(input_tokens=3, output_tokens=7) + background_tasks: list[asyncio.Task[None]] = [] + + async def inner() -> AsyncIterator[str]: + yield serialize_event( + TokenStreamPayload.create(chunk_id=0, token="Hi"), + MEDIA_TYPE_JSON, + ) + + mocker.patch("utils.agents.streaming.consume_query_tokens") + mocker.patch( + "utils.agents.streaming.get_available_quotas", + return_value={"daily": 100}, + ) + mocker.patch( + "utils.agents.streaming.maybe_get_topic_summary", + new=mocker.AsyncMock(return_value=None), + ) + mocker.patch("utils.agents.streaming.store_query_results") + mock_config = mocker.Mock() + mock_config.quota_limiters = [] + mocker.patch("utils.agents.streaming.configuration", mock_config) + + result = [ + event + async for event in generate_agent_response( + inner(), + context, + responses_params, + turn_summary, + background_tasks, + **generate_kwargs, + ) + ] + + end_events = [ + parsed + for event in result + if event.startswith("data: ") + and (parsed := json.loads(event.removeprefix("data: ").strip()))["event"] + == "end" + ] + assert len(end_events) == 1 + assert end_events[0]["data"]["context_status"] == expected_status + @pytest.mark.asyncio async def test_cancelled_persists_interrupted_turn( self, diff --git a/tests/unit/utils/test_conversation_compaction.py b/tests/unit/utils/test_conversation_compaction.py index 29678d691..d3f7baae6 100644 --- a/tests/unit/utils/test_conversation_compaction.py +++ b/tests/unit/utils/test_conversation_compaction.py @@ -121,6 +121,13 @@ def test_should_compact() -> None: ) +def test_compaction_result_context_status() -> None: + """The compacted flag maps to the API context_status value (LCORE-1573).""" + params = _params() + assert cc.CompactionResult(params, compacted=False).context_status == "full" + assert cc.CompactionResult(params, compacted=True).context_status == "summarized" + + # --- apply_compaction --- From a3cc753a83236ccf80d6711efe3ab79a1865b8a5 Mon Sep 17 00:00:00 2001 From: Maxim Svistunov Date: Fri, 14 Aug 2026 07:48:14 +0200 Subject: [PATCH 2/2] LCORE-1573: record two-place context_status split in compaction design doc MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Update the conversation-compaction design doc to match the implemented context_status surface: - Rewrite the "API response changes" section: the field is added to QueryResponse (non-streaming /v1/query) and EndEventData (the streaming SSE end event payload), not to StreamingQueryResponse, which turned out to be a documentation-only class with an empty body — adding a field there would change nothing on the wire, so it is intentionally skipped and only its SSE example is updated. - Replace the stale "src/models/responses.py (now relocated)" row in the key-files table with the two real locations and the docs-only-skip note. - Update the request-flow step 11 note now that LCORE-1573 has landed. --- .../conversation-compaction.md | 28 ++++++++++++++----- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/docs/design/conversation-compaction/conversation-compaction.md b/docs/design/conversation-compaction/conversation-compaction.md index 4098b30db..6ce7155e3 100644 --- a/docs/design/conversation-compaction/conversation-compaction.md +++ b/docs/design/conversation-compaction/conversation-compaction.md @@ -115,7 +115,7 @@ lightspeed-stack 9. Append the completed turn to the conversation items (continuous history, same conversation_id) — Llama Stack did not auto-store it (no conversation param) 10. Release per-conversation lock - 11. Return response (context_status="summarized" when 1573 lands; "full" otherwise) + 11. Return response (context_status="summarized" when compacted; "full" otherwise — LCORE-1573) ``` Note: when no prior summary exists and the request is below the threshold, @@ -237,17 +237,30 @@ This preserves a single continuous conversation identity. The `conversation_id` ## API response changes -Add `context_status` field to `QueryResponse` and `StreamingQueryResponse`: +The `context_status` field is added in two places (LCORE-1573): + +- `QueryResponse` (`src/models/api/responses/successful/query.py`) — the + non-streaming `/v1/query` response body. +- `EndEventData` (`src/models/common/agents/stream_payloads.py`) — the SSE + `end` event payload that streaming `/v1/streaming_query` clients actually + receive on the wire, alongside the analogous `truncated` signal. ``` python -context_status: str = Field( +context_status: ContextStatus = Field( "full", - description="Context status: 'full' (no compaction), " - "'summarized' (older turns summarized).", + description='Context status: "full" (no compaction) or ' + '"summarized" (older turns replaced by a summary)', ) ``` -The existing `truncated` field remains deprecated. +`StreamingQueryResponse` is a documentation-only class with an empty body +(its `openapi_response()` inlines an SSE example string); adding a field +there would change nothing on the wire, so it is intentionally skipped — +only its SSE example is updated to show `context_status` in the `end` event. + +The value maps directly from `CompactionResult.compacted` +(`utils/conversation_compaction.py`): `"summarized"` when True, `"full"` +otherwise. The existing `truncated` field remains deprecated. ## Configuration @@ -307,7 +320,8 @@ Add `compaction` field to the root `Configuration` class. | `src/app/endpoints/streaming_query.py` | Compaction-aware SSE path that emits the `compaction` event before summarizing (R12) | | `src/app/endpoints/a2a.py` | Inline compaction (no SSE event); store the turn on `response.completed` | | `src/app/endpoints/responses.py` | Silent compaction (OpenAI-compatible); store the turn via `_append_previous_response_turn` | -| `src/models/responses.py` (now relocated) | `context_status` field — deferred to LCORE-1573 | +| `src/models/api/responses/successful/query.py` | `context_status` on `QueryResponse` (non-streaming `/v1/query`) — LCORE-1573 | +| `src/models/common/agents/stream_payloads.py` | `context_status` on `EndEventData`, the streaming SSE `end` event payload (`StreamingQueryResponse` is docs-only and intentionally skipped) — LCORE-1573 | | `src/cache/` (all backends) | `ConversationSummary` storage — LCORE-1571 | ## How compaction is invoked