diff --git a/.gitignore b/.gitignore index add55ba88..cfac761d4 100644 --- a/.gitignore +++ b/.gitignore @@ -196,4 +196,7 @@ dev-tools/mcp-mock-server/.certs/ requirements.*.backup # Local run files -local-run.yaml \ No newline at end of file +local-run.yaml + +# Sisyphus planning files +.sisyphus/ \ No newline at end of file diff --git a/docs/responses.md b/docs/responses.md index 148029545..8a8f55342 100644 --- a/docs/responses.md +++ b/docs/responses.md @@ -613,7 +613,7 @@ The `instructions` field on the `/v1/responses` endpoint follows the same server If the server configuration sets `disable_query_system_prompt` to `true`, any request that includes a non-null `instructions` value is rejected with a `422 Unprocessable Entity` error. The error message references the `instructions` field specifically. -This means that even when `instructions` is omitted from the request, the response will always contain a resolved `instructions` value reflecting the server-side default or configured system prompt. +If the server substituted the system prompt, the response sets `instructions` to the placeholder `` instead of echoing the actual prompt. If the client provided their own `instructions`, they are echoed back unchanged. Server-deployed MCP tool definitions are also filtered from the response `tools` array; client-provided tools are preserved. ### Streaming Differences diff --git a/src/app/endpoints/responses.py b/src/app/endpoints/responses.py index 6b9f07f68..03aec7c31 100644 --- a/src/app/endpoints/responses.py +++ b/src/app/endpoints/responses.py @@ -36,6 +36,7 @@ from authorization.middleware import authorize from client import AsyncLlamaStackClientHolder from configuration import configuration +from constants import SUBSTITUTED_INSTRUCTIONS_PLACEHOLDER from log import get_logger from models.config import Action from models.requests import ResponsesRequest @@ -179,9 +180,11 @@ async def responses_endpoint_handler( responses_request = responses_request.model_copy(deep=True) check_configuration_loaded(configuration) + client_instructions = responses_request.instructions responses_request.instructions = get_system_prompt( responses_request.instructions, field_name="instructions" ) + instructions_substituted = client_instructions is None started_at = datetime.now(UTC) user_id = auth[0] @@ -302,6 +305,7 @@ async def responses_endpoint_handler( moderation_result=moderation_result, inline_rag_context=inline_rag_context, filter_server_tools=filter_server_tools, + instructions_substituted=instructions_substituted, ) @@ -314,6 +318,7 @@ async def handle_streaming_response( moderation_result: ShieldModerationResult, inline_rag_context: RAGContext, filter_server_tools: bool = False, + instructions_substituted: bool = False, ) -> StreamingResponse: """Handle streaming response from Responses API. @@ -326,6 +331,7 @@ async def handle_streaming_response( moderation_result: Result of shield moderation check inline_rag_context: Inline RAG context to be used for the response filter_server_tools: Whether to filter server-deployed MCP tool events from the stream + instructions_substituted: Whether the server substituted the instructions Returns: StreamingResponse with SSE-formatted events """ @@ -365,6 +371,7 @@ async def handle_streaming_response( turn_summary=turn_summary, inline_rag_context=inline_rag_context, filter_server_tools=filter_server_tools, + instructions_substituted=instructions_substituted, ) except RuntimeError as e: # library mode wraps 413 into runtime error if is_context_length_error(str(e)): @@ -496,6 +503,45 @@ async def shield_violation_generator( yield "data: [DONE]\n\n" +def _sanitize_response_dict( + response_dict: dict[str, Any], + configured_mcp_labels: set[str], + instructions_substituted: bool = False, +) -> None: + """Sanitize a serialized response object in-place to remove internal details. + + Strips fields that expose server-side implementation details from the + response object before it is forwarded to the client: + + - ``instructions``: when the server substituted its own system prompt + (because the client sent ``None`` or a different value was resolved), + the value is replaced with a placeholder slug to avoid leaking the + actual prompt. When the client provided their own instructions and + they were used as-is, the value is left unchanged. + - ``tools``: server-deployed MCP tool definitions are removed; client- + provided tools (those whose ``server_label`` is not in + ``configured_mcp_labels``) are preserved + + Args: + response_dict: Mutable dict produced by ``model_dump`` on a response + object. Modified in-place. + configured_mcp_labels: Set of ``server_label`` values that identify + server-deployed MCP servers. + instructions_substituted: Whether the server substituted the + instructions (True) or the client provided them (False). + """ + if instructions_substituted: + response_dict["instructions"] = SUBSTITUTED_INSTRUCTIONS_PLACEHOLDER + # else: leave instructions as-is (echo back client's value) + + if tools := response_dict.get("tools"): + response_dict["tools"] = [ + tool + for tool in tools + if tool.get("server_label") not in configured_mcp_labels + ] + + def _is_server_mcp_output_item( item: dict[str, Any], configured_mcp_labels: set[str] ) -> bool: @@ -611,6 +657,7 @@ async def response_generator( turn_summary: TurnSummary, inline_rag_context: RAGContext, filter_server_tools: bool = False, + instructions_substituted: bool = False, ) -> AsyncIterator[str]: """Generate SSE-formatted streaming response with LCORE-enriched events. @@ -622,6 +669,7 @@ async def response_generator( turn_summary: TurnSummary to populate during streaming inline_rag_context: Inline RAG context to be used for the response filter_server_tools: Whether to filter server-deployed MCP tool events from the stream + instructions_substituted: Whether the server substituted the instructions Yields: SSE-formatted strings for streaming events, ending with [DONE] """ @@ -631,9 +679,7 @@ async def response_generator( latest_response_object: Optional[OpenAIResponseObject] = None sequence_number = 0 - configured_mcp_labels = ( - {s.name for s in configuration.mcp_servers} if filter_server_tools else set() - ) + configured_mcp_labels = {s.name for s in configuration.mcp_servers} # Track output indices of server-deployed MCP calls to filter their events server_mcp_output_indices: set[int] = set() @@ -657,6 +703,11 @@ async def response_generator( if "response" in chunk_dict: chunk_dict["response"]["conversation"] = normalized_conv_id + _sanitize_response_dict( + chunk_dict["response"], + configured_mcp_labels, + instructions_substituted, + ) tools = chunk_dict["response"].get("tools") if tools is not None: chunk_dict["response"]["tools"] = ( @@ -794,6 +845,7 @@ async def handle_non_streaming_response( moderation_result: ShieldModerationResult, inline_rag_context: RAGContext, filter_server_tools: bool = False, + instructions_substituted: bool = False, ) -> ResponsesResponse: """Handle non-streaming response from Responses API. @@ -806,6 +858,7 @@ async def handle_non_streaming_response( moderation_result: Result of shield moderation check inline_rag_context: Inline RAG context to be used for the response filter_server_tools: Whether to filter server-deployed MCP tool output + instructions_substituted: Whether the server substituted the instructions Returns: ResponsesResponse with the completed response """ @@ -904,7 +957,11 @@ async def handle_non_streaming_response( skip_userid_check=skip_userid_check, topic_summary=topic_summary, ) + configured_mcp_labels = {s.name for s in configuration.mcp_servers} response_dict = api_response.model_dump(exclude_none=True) + _sanitize_response_dict( + response_dict, configured_mcp_labels, instructions_substituted + ) tools = response_dict.get("tools") if tools is not None: response_dict["tools"] = translate_vector_store_ids_to_user_facing( diff --git a/src/constants.py b/src/constants.py index 6c4a792c4..929d45d40 100644 --- a/src/constants.py +++ b/src/constants.py @@ -229,6 +229,10 @@ DEFAULT_VIOLATION_MESSAGE = "I cannot process this request due to policy restrictions." +# Placeholder slug used in responses when the server substituted its own +# system prompt for the client's instructions. Avoids leaking the actual +# server prompt back to the client. +SUBSTITUTED_INSTRUCTIONS_PLACEHOLDER = "" # Input size limits for API request validation # Maximum character length for the question field in /v1/infer requests (32 KiB) RLSAPI_V1_QUESTION_MAX_LENGTH = 32_768 diff --git a/tests/unit/app/endpoints/test_responses.py b/tests/unit/app/endpoints/test_responses.py index 690683909..d42a443f0 100644 --- a/tests/unit/app/endpoints/test_responses.py +++ b/tests/unit/app/endpoints/test_responses.py @@ -17,14 +17,15 @@ from app.endpoints.responses import ( _is_server_mcp_output_item, + _sanitize_response_dict, _should_filter_mcp_chunk, handle_non_streaming_response, handle_streaming_response, responses_endpoint_handler, ) from configuration import AppConfig -from constants import DEFAULT_SYSTEM_PROMPT -from models.config import Action +from constants import DEFAULT_SYSTEM_PROMPT, SUBSTITUTED_INSTRUCTIONS_PLACEHOLDER +from models.config import Action, ModelContextProtocolServer from models.database.conversations import UserConversation from models.requests import ResponsesRequest from models.responses import ResponsesResponse @@ -1868,3 +1869,296 @@ def test_does_not_filter_non_mcp_event(self, mocker: MockerFixture) -> None: ) is False ) + + +class TestSanitizeResponseDict: + """Unit tests for _sanitize_response_dict.""" + + def test_substituted_instructions_replaced_with_placeholder(self) -> None: + """Test that substituted instructions are replaced with the slug constant.""" + d: dict[str, Any] = {"instructions": "secret server prompt", "model": "m"} + _sanitize_response_dict(d, set(), instructions_substituted=True) + assert d["instructions"] == SUBSTITUTED_INSTRUCTIONS_PLACEHOLDER + + def test_client_instructions_preserved_when_not_substituted(self) -> None: + """Test that client-provided instructions are echoed back unchanged.""" + d: dict[str, Any] = {"instructions": "my custom prompt", "model": "m"} + _sanitize_response_dict(d, set(), instructions_substituted=False) + assert d["instructions"] == "my custom prompt" + + def test_substituted_instructions_set_even_when_absent(self) -> None: + """Test that placeholder is set even when instructions field is missing.""" + d: dict[str, Any] = {"model": "m"} + _sanitize_response_dict(d, set(), instructions_substituted=True) + assert d["instructions"] == SUBSTITUTED_INSTRUCTIONS_PLACEHOLDER + + def test_no_error_when_instructions_absent_and_not_substituted(self) -> None: + """Test that missing instructions field with no substitution does not raise.""" + d: dict[str, Any] = {"model": "m"} + _sanitize_response_dict(d, set(), instructions_substituted=False) + assert "instructions" not in d + + def test_strips_server_mcp_tools(self) -> None: + """Test that server-deployed MCP tools are removed from tools array.""" + d: dict[str, Any] = { + "tools": [ + {"server_label": "server-a", "name": "tool1"}, + {"server_label": "server-b", "name": "tool2"}, + {"name": "client-tool"}, + ] + } + _sanitize_response_dict( + d, {"server-a", "server-b"}, instructions_substituted=False + ) + assert d["tools"] == [{"name": "client-tool"}] + + def test_preserves_client_tools(self) -> None: + """Test that client-provided tools are preserved.""" + d: dict[str, Any] = { + "tools": [ + {"server_label": "server-a", "name": "server-tool"}, + {"name": "client-tool"}, + ] + } + _sanitize_response_dict(d, {"server-a"}, instructions_substituted=False) + assert d["tools"] == [{"name": "client-tool"}] + + def test_no_error_when_tools_absent(self) -> None: + """Test that missing tools field does not raise.""" + d: dict[str, Any] = {"model": "m"} + _sanitize_response_dict(d, {"server-a"}, instructions_substituted=False) + assert "tools" not in d + + def test_empty_configured_mcp_labels_preserves_all_tools(self) -> None: + """Test that empty configured_mcp_labels preserves all tools.""" + d: dict[str, Any] = { + "tools": [ + {"server_label": "server-a", "name": "tool1"}, + {"name": "client-tool"}, + ] + } + _sanitize_response_dict(d, set(), instructions_substituted=False) + assert len(d["tools"]) == 2 + + def test_all_fields_sanitized_together_with_substitution(self) -> None: + """Test that all sanitizations are applied in a single call.""" + d: dict[str, Any] = { + "instructions": "secret prompt", + "model": "google-vertex/publishers/google/models/gemini", + "tools": [ + {"server_label": "mcp-server", "name": "server-tool"}, + {"name": "client-tool"}, + ], + } + _sanitize_response_dict(d, {"mcp-server"}, instructions_substituted=True) + assert d["instructions"] == SUBSTITUTED_INSTRUCTIONS_PLACEHOLDER + assert d["model"] == "google-vertex/publishers/google/models/gemini" + assert d["tools"] == [{"name": "client-tool"}] + + def test_all_fields_sanitized_together_without_substitution(self) -> None: + """Test that client instructions are preserved while tools are still filtered.""" + d: dict[str, Any] = { + "instructions": "client prompt", + "model": "google-vertex/publishers/google/models/gemini", + "tools": [ + {"server_label": "mcp-server", "name": "server-tool"}, + {"name": "client-tool"}, + ], + } + _sanitize_response_dict(d, {"mcp-server"}, instructions_substituted=False) + assert d["instructions"] == "client prompt" + assert d["model"] == "google-vertex/publishers/google/models/gemini" + assert d["tools"] == [{"name": "client-tool"}] + + +class TestMcpEventsFilteredUnconditionally: + """Integration test: MCP events are filtered regardless of X-LCS-Merge-Server-Tools.""" + + @pytest.mark.asyncio + async def test_mcp_events_filtered_without_merge_server_tools_header( + self, + minimal_config: AppConfig, + mocker: MockerFixture, + ) -> None: + """Test MCP streaming events are filtered even without X-LCS-Merge-Server-Tools header.""" + mcp_server = ModelContextProtocolServer( + name="server-a", + provider_id="model-context-protocol", + url="http://mcp.example.com", + ) + mock_config = mocker.Mock() + mock_config.mcp_servers = [mcp_server] + mock_config.quota_limiters = minimal_config.quota_limiters + mock_config.rag_id_mapping = {} + + request = _request_with_model_and_conv("Hi", model="provider/model1") + mock_client = mocker.AsyncMock(spec=AsyncLlamaStackClient) + mock_moderation = mocker.Mock() + mock_moderation.decision = "passed" + + mcp_item = mocker.Mock() + mcp_item.type = "mcp_call" + mcp_item.server_label = "server-a" + + mcp_added_chunk = mocker.Mock() + mcp_added_chunk.type = "response.output_item.added" + mcp_added_chunk.item = mcp_item + mcp_added_chunk.output_index = 0 + mcp_added_chunk.model_dump.return_value = { + "type": "response.output_item.added", + "output_index": 0, + } + + completed_chunk = mocker.Mock() + completed_chunk.type = "response.completed" + completed_chunk.response = mocker.Mock() + completed_chunk.response.id = "r1" + completed_chunk.response.output = [] + completed_chunk.response.usage = mocker.Mock( + input_tokens=1, output_tokens=2, total_tokens=3 + ) + completed_chunk.model_dump.return_value = { + "type": "response.completed", + "response": {"id": "r1"}, + } + + async def mock_stream() -> Any: + yield mcp_added_chunk + yield completed_chunk + + mock_client.responses.create = mocker.AsyncMock(return_value=mock_stream()) + + mocker.patch(f"{MODULE}.configuration", mock_config) + mocker.patch(f"{MODULE}.get_available_quotas", return_value={}) + mocker.patch(f"{MODULE}.extract_token_usage", return_value=mocker.Mock()) + mocker.patch(f"{MODULE}.consume_query_tokens") + mocker.patch(f"{MODULE}.extract_vector_store_ids_from_tools", return_value=[]) + mocker.patch( + f"{MODULE}.build_turn_summary", + return_value=TurnSummary(referenced_documents=[]), + ) + mocker.patch( + f"{MODULE}.get_topic_summary", + new=mocker.AsyncMock(return_value=None), + ) + mocker.patch(f"{MODULE}.store_query_results") + mocker.patch( + f"{MODULE}.normalize_conversation_id", + return_value=VALID_CONV_ID_NORMALIZED, + ) + mock_holder = mocker.Mock() + mock_holder.get_client.return_value = mock_client + mocker.patch(f"{MODULE}.AsyncLlamaStackClientHolder", return_value=mock_holder) + + response = await handle_streaming_response( + client=mock_client, + request=request, + auth=MOCK_AUTH, + input_text="Hi", + started_at=datetime.now(UTC), + moderation_result=mock_moderation, + inline_rag_context=RAGContext(), + filter_server_tools=False, + ) + collected: list[str] = [] + async for part in response.body_iterator: + chunk_str = ( + part.decode("utf-8") + if isinstance(part, bytes) + else (part if isinstance(part, str) else bytes(part).decode("utf-8")) + ) + collected.append(chunk_str) + body = "".join(collected) + assert "response.output_item.added" not in body + assert "response.completed" in body + + @pytest.mark.asyncio + async def test_mcp_events_filtered_with_no_mcp_servers_configured( + self, + minimal_config: AppConfig, + mocker: MockerFixture, + ) -> None: + """Test that non-MCP output_item.added events pass through when no MCP servers configured. + + When no MCP servers are configured, configured_mcp_labels is empty and no events + are filtered. + """ + request = _request_with_model_and_conv("Hi", model="provider/model1") + mock_client = mocker.AsyncMock(spec=AsyncLlamaStackClient) + mock_moderation = mocker.Mock() + mock_moderation.decision = "passed" + + text_item = mocker.Mock() + text_item.type = "message" + + text_added_chunk = mocker.Mock() + text_added_chunk.type = "response.output_item.added" + text_added_chunk.item = text_item + text_added_chunk.output_index = 0 + text_added_chunk.model_dump.return_value = { + "type": "response.output_item.added", + "output_index": 0, + } + + completed_chunk = mocker.Mock() + completed_chunk.type = "response.completed" + completed_chunk.response = mocker.Mock() + completed_chunk.response.id = "r1" + completed_chunk.response.output = [] + completed_chunk.response.usage = mocker.Mock( + input_tokens=1, output_tokens=2, total_tokens=3 + ) + completed_chunk.model_dump.return_value = { + "type": "response.completed", + "response": {"id": "r1"}, + } + + async def mock_stream() -> Any: + yield text_added_chunk + yield completed_chunk + + mock_client.responses.create = mocker.AsyncMock(return_value=mock_stream()) + + mocker.patch(f"{MODULE}.configuration", minimal_config) + mocker.patch(f"{MODULE}.get_available_quotas", return_value={}) + mocker.patch(f"{MODULE}.extract_token_usage", return_value=mocker.Mock()) + mocker.patch(f"{MODULE}.consume_query_tokens") + mocker.patch(f"{MODULE}.extract_vector_store_ids_from_tools", return_value=[]) + mocker.patch( + f"{MODULE}.build_turn_summary", + return_value=TurnSummary(referenced_documents=[]), + ) + mocker.patch( + f"{MODULE}.get_topic_summary", + new=mocker.AsyncMock(return_value=None), + ) + mocker.patch(f"{MODULE}.store_query_results") + mocker.patch( + f"{MODULE}.normalize_conversation_id", + return_value=VALID_CONV_ID_NORMALIZED, + ) + mock_holder = mocker.Mock() + mock_holder.get_client.return_value = mock_client + mocker.patch(f"{MODULE}.AsyncLlamaStackClientHolder", return_value=mock_holder) + + response = await handle_streaming_response( + client=mock_client, + request=request, + auth=MOCK_AUTH, + input_text="Hi", + started_at=datetime.now(UTC), + moderation_result=mock_moderation, + inline_rag_context=RAGContext(), + filter_server_tools=False, + ) + collected: list[str] = [] + async for part in response.body_iterator: + chunk_str = ( + part.decode("utf-8") + if isinstance(part, bytes) + else (part if isinstance(part, str) else bytes(part).decode("utf-8")) + ) + collected.append(chunk_str) + body = "".join(collected) + assert "response.output_item.added" in body + assert "response.completed" in body