diff --git a/deploy/ogx/test.containerfile b/deploy/ogx/test.containerfile index f9e9f3d78..63c413dd5 100644 --- a/deploy/ogx/test.containerfile +++ b/deploy/ogx/test.containerfile @@ -29,6 +29,10 @@ ENV PATH="/opt/app-root/.venv/bin:$PATH" \ # Set HOME directory so OGX uses /opt/app-root/src/.llama ENV HOME="/opt/app-root/src" +# Point OGX at the external provider specs copied with the source tree above +# (e.g. remote::solr_vector_io for OKP RAG). Mirrors the production Containerfile. +ENV EXTERNAL_PROVIDERS_DIR="/opt/app-root/providers/resources/external_providers" + # Create python3 symlink for compatibility RUN ln -sf /usr/bin/python3.12 /usr/bin/python3 diff --git a/docs/devel_doc/openapi.json b/docs/devel_doc/openapi.json index af5a5fbdc..907255878 100644 --- a/docs/devel_doc/openapi.json +++ b/docs/devel_doc/openapi.json @@ -16255,6 +16255,93 @@ "title": "OkpConfiguration", "description": "OKP (Offline Knowledge Portal) provider configuration.\n\nControls provider-specific behaviour for the OKP vector store.\nOnly relevant when ``\"okp\"`` is listed in ``rag.retrieval.inline.sources``\nor ``rag.retrieval.tool.sources``." }, + "OkpFilter": { + "properties": { + "products": { + "items": { + "$ref": "#/components/schemas/OkpProductFilter" + }, + "type": "array", + "title": "Products", + "description": "Product selections to include, OR'd together; each entry scopes its versions to its product.", + "examples": [ + [ + { + "product": "openshift_container_platform", + "versions": [ + "4.16", + "4.17" + ] + } + ], + [ + { + "product": "openshift_container_platform", + "versions": [ + "4.16" + ] + }, + { + "product": "red_hat_enterprise_linux", + "versions": [ + "9", + "10" + ] + } + ] + ] + } + }, + "additionalProperties": false, + "type": "object", + "title": "OkpFilter", + "description": "Transport-neutral, query-time OKP RAG filter.\n\nA domain-oriented product/version filter for the OKP RAG source,\ndeliberately decoupled from any backend filter vocabulary: it uses neither\nSolr's ``fq`` nor the OGX/llama-stack ``{type, key, value}`` grammar. Each\nOKP transport translates it into its own backend form (the RHOKP MCP\n``search`` tool args, or an OGX structured filter for the legacy Solr path),\nso the public interface is unaffected when OGX/Solr is retired.\n\nAttributes:\n products: Product selections to include, OR'd together. Each entry binds\n its versions to its product. An empty list applies no filter." + }, + "OkpProductFilter": { + "properties": { + "product": { + "type": "string", + "title": "Product", + "description": "Exact product identifier (exact match, no wildcards).", + "examples": [ + "openshift_container_platform", + "red_hat_enterprise_linux" + ] + }, + "versions": { + "anyOf": [ + { + "items": { + "type": "string" + }, + "type": "array" + }, + { + "type": "null" + } + ], + "title": "Versions", + "description": "Exact versions of this product to include (exact match, no wildcards). When omitted, the product matches regardless of version.", + "examples": [ + [ + "4.16", + "4.17" + ], + [ + "9", + "10" + ] + ] + } + }, + "additionalProperties": false, + "type": "object", + "required": [ + "product" + ], + "title": "OkpProductFilter", + "description": "A single product selection, with its versions scoped to that product.\n\nVersions live under their product so an invalid cross-pairing (e.g. a\nversion that belongs to a different product) is structurally\nunrepresentable. All values are matched exactly (no wildcards).\n\nAttributes:\n product: Exact product identifier to filter on (e.g.\n ``openshift_container_platform``).\n versions: Exact versions of this product to include. When None or empty,\n the product matches regardless of version." + }, "OpenAIResponseAnnotationCitation": { "properties": { "type": { @@ -18953,6 +19040,30 @@ } } ] + }, + "okp": { + "anyOf": [ + { + "$ref": "#/components/schemas/OkpFilter" + }, + { + "type": "null" + } + ], + "description": "Transport-neutral OKP RAG filter: exact-match product selections (each with its own versions) applied by whichever OKP transport is active (RHOKP MCP or the legacy Solr path).", + "examples": [ + { + "products": [ + { + "product": "openshift_container_platform", + "versions": [ + "4.16", + "4.17" + ] + } + ] + } + ] } }, "additionalProperties": false, @@ -18961,7 +19072,7 @@ "query" ], "title": "QueryRequest", - "description": "Model representing a request for the LLM (Language Model).\n\nAttributes:\n query: The query string.\n conversation_id: The optional conversation ID (UUID).\n provider: The optional provider.\n model: The optional model.\n system_prompt: The optional system prompt.\n attachments: The optional attachments.\n no_tools: Whether to bypass all tools and MCP servers (default: False).\n generate_topic_summary: Whether to generate topic summary for new conversations.\n media_type: The optional media type for response format (application/json or text/plain).\n vector_store_ids: The optional list of specific vector store IDs to query for RAG.\n shield_ids: The optional list of configured shield names to apply.\n solr: Optional Solr inline RAG options (mode, filters) or legacy filter-only dict.", + "description": "Model representing a request for the LLM (Language Model).\n\nAttributes:\n query: The query string.\n conversation_id: The optional conversation ID (UUID).\n provider: The optional provider.\n model: The optional model.\n system_prompt: The optional system prompt.\n attachments: The optional attachments.\n no_tools: Whether to bypass all tools and MCP servers (default: False).\n generate_topic_summary: Whether to generate topic summary for new conversations.\n media_type: The optional media type for response format (application/json or text/plain).\n vector_store_ids: The optional list of specific vector store IDs to query for RAG.\n shield_ids: The optional list of configured shield names to apply.\n solr: Optional Solr inline RAG options (mode, filters) or legacy filter-only dict.\n okp: Optional transport-neutral OKP RAG filter (product, product_version).", "examples": [ { "attachments": [ @@ -20398,6 +20509,16 @@ "type": "null" } ] + }, + "okp": { + "anyOf": [ + { + "$ref": "#/components/schemas/OkpFilter" + }, + { + "type": "null" + } + ] } }, "additionalProperties": false, @@ -20406,7 +20527,7 @@ "input" ], "title": "ResponsesRequest", - "description": "Model representing a request for the Responses API following LCORE specification.\n\nAttributes:\n input: Input text or structured input items containing the query.\n model: Model identifier in format \"provider/model\". Auto-selected if not provided.\n conversation: Conversation ID linking to an existing conversation. Accepts both\n OpenAI and LCORE formats. Mutually exclusive with previous_response_id.\n include: Explicitly specify output item types that are excluded by default but\n should be included in the response.\n instructions: System instructions or guidelines provided to the model (acts as\n the system prompt).\n max_infer_iters: Maximum number of inference iterations the model can perform.\n max_output_tokens: Maximum number of tokens allowed in the response.\n max_tool_calls: Maximum number of tool calls allowed in a single response.\n metadata: Custom metadata dictionary with key-value pairs for tracking or logging.\n parallel_tool_calls: Whether the model can make multiple tool calls in parallel.\n previous_response_id: Identifier of the previous response in a multi-turn\n conversation. Mutually exclusive with conversation.\n prompt: Prompt object containing a template with variables for dynamic\n substitution.\n reasoning: Reasoning configuration for the response.\n safety_identifier: Safety identifier for the response.\n store: Whether to store the response in conversation history. Defaults to True.\n stream: Whether to stream the response as it is generated. Defaults to False.\n temperature: Sampling temperature controlling randomness (typically 0.0\u20132.0).\n text: Text response configuration specifying output format constraints (JSON\n schema, JSON object, or plain text).\n tool_choice: Tool selection strategy (\"auto\", \"required\", \"none\", or specific\n tool configuration).\n tools: List of tools available to the model (file search, web search, function\n calls, MCP tools). Defaults to all tools available to the model.\n generate_topic_summary: LCORE-specific flag indicating whether to generate a\n topic summary for new conversations. Defaults to True.\n shield_ids: LCORE-specific list of configured shield names to apply.\n If None, all configured shields are used.\n solr: Optional Solr inline RAG options (mode, filters) or legacy filter-only dict.", + "description": "Model representing a request for the Responses API following LCORE specification.\n\nAttributes:\n input: Input text or structured input items containing the query.\n model: Model identifier in format \"provider/model\". Auto-selected if not provided.\n conversation: Conversation ID linking to an existing conversation. Accepts both\n OpenAI and LCORE formats. Mutually exclusive with previous_response_id.\n include: Explicitly specify output item types that are excluded by default but\n should be included in the response.\n instructions: System instructions or guidelines provided to the model (acts as\n the system prompt).\n max_infer_iters: Maximum number of inference iterations the model can perform.\n max_output_tokens: Maximum number of tokens allowed in the response.\n max_tool_calls: Maximum number of tool calls allowed in a single response.\n metadata: Custom metadata dictionary with key-value pairs for tracking or logging.\n parallel_tool_calls: Whether the model can make multiple tool calls in parallel.\n previous_response_id: Identifier of the previous response in a multi-turn\n conversation. Mutually exclusive with conversation.\n prompt: Prompt object containing a template with variables for dynamic\n substitution.\n reasoning: Reasoning configuration for the response.\n safety_identifier: Safety identifier for the response.\n store: Whether to store the response in conversation history. Defaults to True.\n stream: Whether to stream the response as it is generated. Defaults to False.\n temperature: Sampling temperature controlling randomness (typically 0.0\u20132.0).\n text: Text response configuration specifying output format constraints (JSON\n schema, JSON object, or plain text).\n tool_choice: Tool selection strategy (\"auto\", \"required\", \"none\", or specific\n tool configuration).\n tools: List of tools available to the model (file search, web search, function\n calls, MCP tools). Defaults to all tools available to the model.\n generate_topic_summary: LCORE-specific flag indicating whether to generate a\n topic summary for new conversations. Defaults to True.\n shield_ids: LCORE-specific list of configured shield names to apply.\n If None, all configured shields are used.\n solr: Optional Solr inline RAG options (mode, filters) or legacy filter-only dict.\n okp: Optional transport-neutral OKP RAG filter (product selections with versions).", "examples": [ { "generate_topic_summary": true, diff --git a/docs/user_doc/rag_guide.md b/docs/user_doc/rag_guide.md index ee6c43cb5..09a8483bb 100644 --- a/docs/user_doc/rag_guide.md +++ b/docs/user_doc/rag_guide.md @@ -395,6 +395,31 @@ Example: } ``` +The `solr` field and its `fq`/structured filters are Solr-specific. For a +backend-neutral, forward-compatible product/version filter that works across +**both** OKP transports (the RHOKP MCP server and the legacy Solr path), use the +request field **`okp`** instead. It carries a list of product selections, each +scoping its own exact-match versions (no wildcards), so an invalid +product/version pairing is not representable. Product/version filtering is +query-time only; there is no launch-time product/version configuration. + +```json +{ + "query": "How do I configure routes?", + "okp": { + "products": [ + { "product": "openshift_container_platform", "versions": ["4.16", "4.17"] } + ] + } +} +``` + +Multiple products are OR'd; a product with no `versions` matches regardless of +version. On the MCP transport each (product, version) pair becomes one search and +the results are merged, deduplicated, and capped at `rag.okp.max_chunks`. Prefer +`okp` over `solr` for product/version filtering; `solr` remains for Solr-specific +needs during the OGX/Solr transition. + **Prerequisites:** - The OKP server must be running and accessible at the URL given in `rag.okp.rhokp_url` (or `${env.RH_SERVER_OKP}`). diff --git a/src/app/endpoints/a2a.py b/src/app/endpoints/a2a.py index 3bd6db1c2..5ec20bdb2 100644 --- a/src/app/endpoints/a2a.py +++ b/src/app/endpoints/a2a.py @@ -456,6 +456,7 @@ async def _process_task_streaming( # pylint: disable=too-many-locals ), shield_ids=None, solr=None, + okp=None, ) # Get LLM client and select model diff --git a/src/app/endpoints/query.py b/src/app/endpoints/query.py index 33865574a..88760aa2a 100644 --- a/src/app/endpoints/query.py +++ b/src/app/endpoints/query.py @@ -220,6 +220,7 @@ async def _handle_query_with_tracing( query_request.query, query_request.vector_store_ids, query_request.solr, + query_request.okp, ) # Prepare API request parameters diff --git a/src/app/endpoints/responses.py b/src/app/endpoints/responses.py index ee554ef79..23b3b53b9 100644 --- a/src/app/endpoints/responses.py +++ b/src/app/endpoints/responses.py @@ -688,6 +688,7 @@ async def handle_responses_with_tracing( # pylint: disable=too-many-locals input_text, vector_store_ids, original_request.solr, + original_request.okp, ) if moderation_result.decision == "passed": updated_request.input = append_inline_rag_context_to_responses_input( diff --git a/src/app/endpoints/streaming_query.py b/src/app/endpoints/streaming_query.py index 483639e25..3dbe652cd 100644 --- a/src/app/endpoints/streaming_query.py +++ b/src/app/endpoints/streaming_query.py @@ -265,6 +265,7 @@ async def _handle_streaming_query_with_tracing( # pylint: disable=too-many-loca query_request.query, query_request.vector_store_ids, query_request.solr, + query_request.okp, ) # Prepare API request parameters diff --git a/src/configuration.py b/src/configuration.py index 4e2750d61..cd0f02b35 100644 --- a/src/configuration.py +++ b/src/configuration.py @@ -1,6 +1,9 @@ """Configuration loader.""" +import asyncio +import time from typing import Any, Optional, Self +from urllib.parse import urljoin import yaml @@ -643,6 +646,22 @@ def inline_solr_enabled(self) -> bool: raise LogicError("logic error: configuration is not loaded") return constants.OKP_RAG_ID in self._configuration.rag.retrieval.inline.sources + @property + def okp_inline_enabled(self) -> bool: + """Return whether OKP is included in the inline RAG list. + + Transport-agnostic alias of :attr:`inline_solr_enabled`: it is True when + ``"okp"`` appears in ``rag.retrieval.inline.sources`` regardless of + whether the Solr or MCP transport is selected. + + Returns: + bool: True if 'okp' appears in rag.inline, False otherwise. + + Raises: + LogicError: If the configuration has not been loaded. + """ + return self.inline_solr_enabled + def resolve_index_name( self, vector_store_id: str, rag_id_mapping: Optional[dict[str, str]] = None ) -> str: @@ -665,3 +684,109 @@ def resolve_index_name( configuration: AppConfig = AppConfig() + + +def okp_mcp_endpoint_url() -> str: + """Return the RHOKP MCP endpoint URL derived from ``rag.okp.rhokp_url``. + + The RHOKP MCP server is always served at the ``/mcp`` path of the RHOKP + base URL, so the endpoint is derived rather than configured separately. When + ``rhokp_url`` is unset, the constant default base URL is used. + + Returns: + str: The absolute MCP endpoint URL (``/mcp``). + """ + okp = configuration.okp + base = ( + str(okp.rhokp_url) + if okp.rhokp_url is not None + else constants.RH_SERVER_OKP_DEFAULT_URL + ) + return urljoin(base, "/mcp") + + +# Cached OKP transport-capability probe result (per process), with the +# monotonic time it was taken. The result is trusted for +# ``OKP_MCP_PROBE_TTL_SECONDS``; after that it is re-probed so a RHOKP instance +# upgraded (or downgraded) while LCORE runs is picked up without a restart. +# None -> not yet probed +# True -> RHOKP advertises the MCP search tool; use the MCP transport +# False -> RHOKP is not MCP-capable (or was unreachable); use the Solr transport +_okp_mcp_available: Optional[bool] = None # pylint: disable=invalid-name +_okp_mcp_probed_at: float = 0.0 # pylint: disable=invalid-name +_okp_mcp_probe_lock = asyncio.Lock() + + +def _okp_mcp_probe_fresh(now: float) -> bool: + """Return whether the cached probe result is still within its TTL. + + Parameters: + now: Current monotonic timestamp. + + Returns: + bool: True if a probe result exists and has not yet expired, False if + there is no cached result or the TTL has elapsed (a re-probe is due). + """ + return ( + _okp_mcp_available is not None + and (now - _okp_mcp_probed_at) < constants.OKP_MCP_PROBE_TTL_SECONDS + ) + + +async def okp_mcp_available() -> bool: + """Return whether the OKP-over-MCP RAG transport should be used. + + At launch time the Solr transport is always assumed (the OGX ``vector_io`` + provider is wired unconditionally). At query time this probes the RHOKP + ``/mcp`` endpoint and caches the result for ``OKP_MCP_PROBE_TTL_SECONDS`` so + that a RHOKP instance which is MCP-incapable or was down does not incur a + probe timeout on every request, while still being re-probed periodically so + an upgraded RHOKP is adopted without restarting LCORE. The Solr transport is + used whenever this returns False, including when the probe fails. + + Returns: + bool: True if the RHOKP MCP endpoint advertises the search tool, False + otherwise (not MCP-capable, unreachable, or recently marked unavailable + by :func:`mark_okp_mcp_unavailable` and still within the TTL). + """ + global _okp_mcp_available, _okp_mcp_probed_at # pylint: disable=global-statement + if _okp_mcp_probe_fresh(time.monotonic()): + return bool(_okp_mcp_available) + async with _okp_mcp_probe_lock: + if _okp_mcp_probe_fresh(time.monotonic()): + return bool(_okp_mcp_available) + # Imported lazily to avoid a circular import at module load time. + from pydantic_ai_lightspeed.retrieval.okp_mcp._client import ( # pylint: disable=import-outside-toplevel + probe_okp_mcp, + ) + + _okp_mcp_available = await probe_okp_mcp( + okp_mcp_endpoint_url(), + tool_name=constants.OKP_MCP_DEFAULT_TOOL_NAME, + timeout=constants.OKP_MCP_PROBE_TIMEOUT_SECONDS, + ) + _okp_mcp_probed_at = time.monotonic() + return _okp_mcp_available + + +def mark_okp_mcp_unavailable() -> None: + """Fall back to Solr after a hard MCP failure at query time. + + Marks the transport unavailable and stamps the current time, so subsequent + requests use Solr without re-attempting MCP until the TTL elapses, at which + point :func:`okp_mcp_available` re-probes and can fail forward to MCP again. + """ + global _okp_mcp_available, _okp_mcp_probed_at # pylint: disable=global-statement + _okp_mcp_available = False + _okp_mcp_probed_at = time.monotonic() + + +def reset_okp_mcp_probe() -> None: + """Reset the cached MCP-capability probe result. + + Intended for tests and for reloading configuration; the next call to + :func:`okp_mcp_available` re-probes the RHOKP endpoint. + """ + global _okp_mcp_available, _okp_mcp_probed_at # pylint: disable=global-statement + _okp_mcp_available = None + _okp_mcp_probed_at = 0.0 diff --git a/src/constants.py b/src/constants.py index 3c6c6355d..f3da6ac2c 100644 --- a/src/constants.py +++ b/src/constants.py @@ -270,6 +270,23 @@ # Special RAG ID that activates the OKP provider when listed in rag.inline or rag.tool OKP_RAG_ID: Final[str] = "okp" +# OKP-over-MCP RAG constants +# The OKP RAG source has two interchangeable transports. At launch the Solr +# vector_io provider is always assumed; at query time the RHOKP MCP endpoint +# (always served at the "/mcp" path of rag.okp.rhokp_url) is probed once and +# preferred when available, falling back to Solr otherwise. The "okp" source id +# activates OKP either way; only the transport changes. +# Default MCP tool name used for OKP hybrid search. +OKP_MCP_DEFAULT_TOOL_NAME: Final[str] = "search" +# Maximum rows the RHOKP MCP search tool accepts (server clamps to 1..=20). +OKP_MCP_MAX_ROWS: Final[int] = 20 +# Timeout (seconds) for the RHOKP MCP capability probe at query time. +OKP_MCP_PROBE_TIMEOUT_SECONDS: Final[float] = 5.0 +# How long a probe result is trusted before re-probing. Bounds probe cost when +# RHOKP is not MCP-capable, while still adopting an upgraded RHOKP without a +# restart ("fail forward"). +OKP_MCP_PROBE_TTL_SECONDS: Final[float] = 60.0 + # OpenTelemetry anonymization configuration # Environment variable for HMAC secret used to anonymize sensitive trace data OTEL_ANONYMIZATION_SECRET_ENV_VAR: Final[str] = "OTEL_ANONYMIZATION_SECRET" diff --git a/src/models/api/requests/query.py b/src/models/api/requests/query.py index 8b6e7b83d..db05cbb8f 100644 --- a/src/models/api/requests/query.py +++ b/src/models/api/requests/query.py @@ -5,7 +5,7 @@ from pydantic import BaseModel, Field, field_validator, model_validator from constants import MEDIA_TYPE_JSON, MEDIA_TYPE_TEXT -from models.common.query import Attachment, SolrVectorSearchRequest +from models.common.query import Attachment, OkpFilter, SolrVectorSearchRequest from utils import suid @@ -25,6 +25,7 @@ class QueryRequest(BaseModel): vector_store_ids: The optional list of specific vector store IDs to query for RAG. shield_ids: The optional list of configured shield names to apply. solr: Optional Solr inline RAG options (mode, filters) or legacy filter-only dict. + okp: Optional transport-neutral OKP RAG filter (product, product_version). """ query: str = Field( @@ -122,6 +123,25 @@ class QueryRequest(BaseModel): ], ) + okp: Optional[OkpFilter] = Field( + None, + description=( + "Transport-neutral OKP RAG filter: exact-match product selections " + "(each with its own versions) applied by whichever OKP transport is " + "active (RHOKP MCP or the legacy Solr path)." + ), + examples=[ + { + "products": [ + { + "product": "openshift_container_platform", + "versions": ["4.16", "4.17"], + } + ] + }, + ], + ) + # provides examples for /docs endpoint model_config = { "extra": "forbid", diff --git a/src/models/api/requests/responses_openai.py b/src/models/api/requests/responses_openai.py index 809b6963f..43bbdd97e 100644 --- a/src/models/api/requests/responses_openai.py +++ b/src/models/api/requests/responses_openai.py @@ -18,7 +18,7 @@ from pydantic import BaseModel, field_validator, model_validator from constants import RESPONSES_REQUEST_MAX_SIZE -from models.common.query import SolrVectorSearchRequest +from models.common.query import OkpFilter, SolrVectorSearchRequest from models.common.responses.types import IncludeParameter, InputTool, ResponseInput from utils import suid @@ -60,6 +60,7 @@ class ResponsesRequest(BaseModel): shield_ids: LCORE-specific list of configured shield names to apply. If None, all configured shields are used. solr: Optional Solr inline RAG options (mode, filters) or legacy filter-only dict. + okp: Optional transport-neutral OKP RAG filter (product selections with versions). """ input: ResponseInput @@ -86,6 +87,7 @@ class ResponsesRequest(BaseModel): generate_topic_summary: Optional[bool] = True shield_ids: Optional[list[str]] = None solr: Optional[SolrVectorSearchRequest] = None + okp: Optional[OkpFilter] = None model_config = { "extra": "forbid", diff --git a/src/models/common/query.py b/src/models/common/query.py index 3facfbf9c..0259b77ce 100644 --- a/src/models/common/query.py +++ b/src/models/common/query.py @@ -134,6 +134,70 @@ def validate_image_attachment(self) -> Self: } +class OkpProductFilter(BaseModel): + """A single product selection, with its versions scoped to that product. + + Versions live under their product so an invalid cross-pairing (e.g. a + version that belongs to a different product) is structurally + unrepresentable. All values are matched exactly (no wildcards). + + Attributes: + product: Exact product identifier to filter on (e.g. + ``openshift_container_platform``). + versions: Exact versions of this product to include. When None or empty, + the product matches regardless of version. + """ + + model_config = ConfigDict(extra="forbid") + + product: str = Field( + description="Exact product identifier (exact match, no wildcards).", + examples=["openshift_container_platform", "red_hat_enterprise_linux"], + ) + versions: Optional[list[str]] = Field( + None, + description=( + "Exact versions of this product to include (exact match, no " + "wildcards). When omitted, the product matches regardless of " + "version." + ), + examples=[["4.16", "4.17"], ["9", "10"]], + ) + + +class OkpFilter(BaseModel): + """Transport-neutral, query-time OKP RAG filter. + + A domain-oriented product/version filter for the OKP RAG source, + deliberately decoupled from any backend filter vocabulary: it uses neither + Solr's ``fq`` nor the OGX/llama-stack ``{type, key, value}`` grammar. Each + OKP transport translates it into its own backend form (the RHOKP MCP + ``search`` tool args, or an OGX structured filter for the legacy Solr path), + so the public interface is unaffected when OGX/Solr is retired. + + Attributes: + products: Product selections to include, OR'd together. Each entry binds + its versions to its product. An empty list applies no filter. + """ + + model_config = ConfigDict(extra="forbid") + + products: list[OkpProductFilter] = Field( + default_factory=list, + description=( + "Product selections to include, OR'd together; each entry scopes " + "its versions to its product." + ), + examples=[ + [{"product": "openshift_container_platform", "versions": ["4.16", "4.17"]}], + [ + {"product": "openshift_container_platform", "versions": ["4.16"]}, + {"product": "red_hat_enterprise_linux", "versions": ["9", "10"]}, + ], + ], + ) + + class SolrVectorSearchRequest(BaseModel): """LCORE Solr inline RAG options for vector_io.query (mode and provider filters). diff --git a/src/ogx_configuration.py b/src/ogx_configuration.py index bf6b06457..f39c7dd5e 100644 --- a/src/ogx_configuration.py +++ b/src/ogx_configuration.py @@ -1334,6 +1334,9 @@ def synthesize_configuration( # pylint: disable=too-many-locals "tool": retrieval.get("tool", {}).get("sources", []), } okp_config = rag_section.get("okp", {}) + # The Solr vector_io provider is always wired at launch. The RHOKP MCP + # transport is selected at query time (utils.vector_search._fetch_okp) and + # falls back to this Solr provider, so it must always be present. enrich_solr(ogx_config, rag_config_for_solr, okp_config) enrich_vector_store(ogx_config, lcs_config.get("vector_store")) @@ -1501,6 +1504,8 @@ def generate_configuration( "tool": retrieval.get("tool", {}).get("sources", []), } okp_config = rag_section.get("okp", {}) + # Solr is always wired; the MCP transport is chosen at query time with Solr + # as the fallback (see synthesize()). enrich_solr(ogx_config, rag_config_for_solr, okp_config) dedupe_providers_vector_io(ogx_config) diff --git a/src/pydantic_ai_lightspeed/retrieval/README.md b/src/pydantic_ai_lightspeed/retrieval/README.md new file mode 100644 index 000000000..7ac5eb75f --- /dev/null +++ b/src/pydantic_ai_lightspeed/retrieval/README.md @@ -0,0 +1,25 @@ +# retrieval + +Pydantic-AI-side RAG context retrievers. Retrievers here fetch RAG context +without going through the OGX `vector_io` client, and emit the backend-neutral +`RAGChunk` / `ReferencedDocument` models from +`models.common.turn_summary` so any retriever produces the same +response/transcript shape. + +## Subpackages + +- `okp_mcp/` — OKP RAG over the RHOKP MCP server. The Pydantic-AI analog of the + OGX/Solr OKP `vector_io` path. + +## Relationship to the Solr OKP path + +The Solr OKP retriever (`utils/vector_search.py:_fetch_okp_rag`) and the MCP +retriever are two transports for the same `"okp"` RAG source, selected per +request by `utils/vector_search.py:_fetch_okp`. The Solr `vector_io` provider is +always wired at launch; at query time the RHOKP MCP transport is preferred +whenever `configuration.okp_mcp_available()` is True (the endpoint, always at +`rhokp_url/mcp`, is probed and cached with a TTL so an upgraded RHOKP is adopted +without restart), and the Solr path serves as the fallback. Both return +`tuple[list[RAGChunk], list[ReferencedDocument]]` and plug into the same +`build_rag_context` merge/rerank/format pipeline (Option A in the design doc). + diff --git a/src/pydantic_ai_lightspeed/retrieval/__init__.py b/src/pydantic_ai_lightspeed/retrieval/__init__.py new file mode 100644 index 000000000..aea0da0a9 --- /dev/null +++ b/src/pydantic_ai_lightspeed/retrieval/__init__.py @@ -0,0 +1,14 @@ +"""Pydantic-AI-side RAG context retrievers. + +This package holds retrievers that fetch RAG context on the Pydantic-AI side of +the stack (i.e. without going through the OGX ``vector_io`` client). Retrievers +emit the backend-neutral chunk/document models +(:class:`models.common.turn_summary.RAGChunk` / +:class:`models.common.turn_summary.ReferencedDocument`) so any retriever feeds +the same response/transcript shape. + +Currently exposes: + - :mod:`pydantic_ai_lightspeed.retrieval.okp_mcp`: OKP RAG over the RHOKP + MCP server (the Pydantic-AI analog of the OGX/Solr OKP ``vector_io`` + path). +""" diff --git a/src/pydantic_ai_lightspeed/retrieval/okp_mcp/README.md b/src/pydantic_ai_lightspeed/retrieval/okp_mcp/README.md new file mode 100644 index 000000000..e952f96ae --- /dev/null +++ b/src/pydantic_ai_lightspeed/retrieval/okp_mcp/README.md @@ -0,0 +1,53 @@ +# okp_mcp + +OKP RAG retriever over the RHOKP MCP server. + +The RHOKP container ships an MCP server that encapsulates embeddings and Solr +querying server-side. This retriever connects to it over streamable HTTP, +calls its search tool with the raw user query, and maps the returned documents +to `RAGChunk` / `ReferencedDocument`. + +Compared to the OGX/Solr OKP `vector_io` path, there is **no** local embedding +model, **no** `vector_io` provider, and **no** OGX `run.yaml` enrichment — the +MCP server does that work. + +## Files + +- `_client.py` — thin wrapper over pydantic-ai's `MCPToolset` + (`direct_call_tool`) that performs a single, programmatic search call. No + agent/LLM is involved. +- `_provider.py` — `OkpMcpRetriever`: `fetch(query) -> (list[RAGChunk], + list[ReferencedDocument])`. Builds itself from configuration via + `OkpMcpRetriever.from_configuration()`. + +## RHOKP MCP `search` tool contract + +- Input: `{"query": str, "rows": int}` (`rows` clamped server-side to 1..20), + plus optional `product` / `product_version` scalars driven by the query-time + `okp` request filter. +- Output: `{"response": {"numFound": int, "docs": [SolrDoc]}}` where each + `SolrDoc` may carry `chunk` (text), `score`, `title`, `doc_id`, + `online_source_url`, `source_path`, `product`, `product_version`. + +## Configuration + +There is no MCP-specific configuration block. The config is the same as the +pre-MCP OKP config; the MCP endpoint is always derived from `rhokp_url` (the +RHOKP MCP server is always served at its `/mcp` path): + +```yaml +rag: + okp: + rhokp_url: ${env.RH_SERVER_OKP} # base URL; MCP endpoint = rhokp_url/mcp + offline: true + retrieval: + inline: + sources: ["okp"] # "okp" activates OKP +``` + +Transport selection is automatic and per-request. The Solr `vector_io` +transport is always wired at launch; at query time the MCP transport is +preferred whenever `configuration.okp_mcp_available()` is True (the endpoint is +probed and TTL-cached so an upgraded RHOKP is adopted without a restart), and +the Solr path serves as the fallback when MCP is unavailable or hard-fails. + diff --git a/src/pydantic_ai_lightspeed/retrieval/okp_mcp/__init__.py b/src/pydantic_ai_lightspeed/retrieval/okp_mcp/__init__.py new file mode 100644 index 000000000..1f116e7fd --- /dev/null +++ b/src/pydantic_ai_lightspeed/retrieval/okp_mcp/__init__.py @@ -0,0 +1,19 @@ +"""OKP RAG retriever over the RHOKP MCP server. + +The RHOKP container ships an MCP server that encapsulates embeddings and Solr +querying server-side. This retriever connects to that server over streamable +HTTP (via pydantic-ai's :class:`~pydantic_ai.mcp.MCPToolset`), calls its search +tool with the raw user query, and maps the returned documents to the +backend-neutral :class:`~models.common.turn_summary.RAGChunk` / +:class:`~models.common.turn_summary.ReferencedDocument` models. + +Unlike the OGX/Solr OKP path, this retriever needs no client-side embedding +model and does no OGX ``run.yaml`` enrichment: the MCP server does the work. The +Solr transport is always wired at launch and serves as the query-time fallback, +selected whenever :func:`configuration.okp_mcp_available` returns False (the +RHOKP endpoint is not MCP-capable or was unreachable). +""" + +from pydantic_ai_lightspeed.retrieval.okp_mcp._provider import OkpMcpRetriever + +__all__ = ["OkpMcpRetriever"] diff --git a/src/pydantic_ai_lightspeed/retrieval/okp_mcp/_client.py b/src/pydantic_ai_lightspeed/retrieval/okp_mcp/_client.py new file mode 100644 index 000000000..df6ddecdf --- /dev/null +++ b/src/pydantic_ai_lightspeed/retrieval/okp_mcp/_client.py @@ -0,0 +1,149 @@ +"""Thin MCP client wrapper for OKP RAG retrieval. + +Wraps pydantic-ai's :class:`~pydantic_ai.mcp.MCPToolset` (FastMCP-backed, +streamable HTTP) to perform a single, programmatic search-tool call against the +RHOKP MCP server. No agent or LLM is involved: this is a pre-run retriever, so +the tool is invoked directly rather than exposed to a model. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any, Optional + +from pydantic_ai.mcp import MCPToolset + +from log import get_logger + +logger = get_logger(__name__) + + +class OkpMcpUnavailableError(RuntimeError): + """Raised when every RHOKP MCP search attempt fails for a single request. + + Signals :func:`utils.vector_search.build_rag_context` that the MCP transport + is unusable for this request and it should fall back to the Solr transport. + """ + + +async def call_okp_search( # pylint: disable=too-many-arguments,too-many-positional-arguments + url: str, + tool_name: str, + query: str, + rows: int, + headers: Optional[dict[str, str]] = None, + timeout: Optional[float] = None, + products: Optional[list[dict[str, Any]]] = None, +) -> dict[str, Any]: + """Call the RHOKP MCP search tool and return its structured result. + + Opens a short-lived streamable-HTTP MCP session, invokes ``tool_name`` with + ``{"query": query, "rows": rows}`` (plus a structured ``products`` filter + when supplied), and returns the tool's structured content. The session is + opened and closed by + :meth:`~pydantic_ai.mcp.MCPToolset.direct_call_tool`. + + The RHOKP MCP ``search`` tool accepts a structured, Solr-``fq``-analogous + product filter and builds the query-side filter itself, so a + multi-product/multi-version selection is expressed in a single tool call + rather than fanned out into one call per (product, version) pair. + + Parameters: + url: RHOKP MCP endpoint (streamable HTTP), e.g. ``http://host:8080/mcp``. + tool_name: Name of the MCP search tool to call (e.g. ``search``). + query: Raw user query string. + rows: Maximum number of results to request (server clamps to 1..20). + headers: Optional static request headers (e.g. authorization). + timeout: Optional per-request timeout in seconds for init and read. + products: Optional structured product filter, a list of + ``{"product": str, "versions": [str, ...]}`` entries (products + OR-combined; versions within a product OR-combined). Omitted from the + tool args when None or empty, in which case the search spans all + products. + + Returns: + The tool's structured content as a dict, e.g. + ``{"response": {"docs": [...], "numFound": N}}``. Returns an empty dict + when the server returns non-mapping content (nothing usable to map). + + Raises: + Exception: Propagates transport- and tool-level errors from the MCP + client (``tool_error_behavior="error"``). Callers are expected to + handle failures and degrade gracefully. + """ + toolset_kwargs: dict[str, Any] = {"tool_error_behavior": "error"} + if headers: + toolset_kwargs["headers"] = headers + if timeout is not None: + toolset_kwargs["init_timeout"] = timeout + toolset_kwargs["read_timeout"] = timeout + + tool_args: dict[str, Any] = {"query": query, "rows": rows} + if products: + tool_args["products"] = products + + toolset = MCPToolset(url, **toolset_kwargs) + result = await toolset.direct_call_tool(tool_name, tool_args) + + if isinstance(result, Mapping): + return dict(result) + + logger.warning( + "OKP MCP tool %r returned non-mapping result of type %s; ignoring", + tool_name, + type(result).__name__, + ) + return {} + + +async def probe_okp_mcp( + url: str, + tool_name: str, + headers: Optional[dict[str, str]] = None, + timeout: Optional[float] = None, +) -> bool: + """Probe an RHOKP endpoint for OKP-over-MCP capability. + + Opens a short-lived streamable-HTTP MCP session against ``url`` and lists + the advertised tools, checking that the OKP search tool is present. Called at + query time (TTL-cached by :func:`configuration.okp_mcp_available`) to + auto-select the OKP RAG transport (MCP vs. the legacy Solr ``vector_io`` + path) without an explicit configuration flag. + + Parameters: + url: Candidate RHOKP MCP endpoint (streamable HTTP), e.g. + ``http://host:8080/mcp``. + tool_name: Name of the search tool the RHOKP MCP server must advertise + for the endpoint to count as MCP-capable. + headers: Optional static request headers (e.g. authorization). + timeout: Optional timeout in seconds for the initialization and read. + + Returns: + True when the endpoint speaks MCP and advertises ``tool_name``; False on + any transport, protocol, or timeout error, or when the tool is absent. + Never raises: probe failures degrade to the Solr transport. + """ + toolset_kwargs: dict[str, Any] = {} + if headers: + toolset_kwargs["headers"] = headers + if timeout is not None: + toolset_kwargs["init_timeout"] = timeout + toolset_kwargs["read_timeout"] = timeout + + try: + toolset = MCPToolset(url, **toolset_kwargs) + tools = await toolset.list_tools() + except Exception as exc: # pylint: disable=broad-exception-caught + logger.info( + "OKP MCP probe of %r failed (%s); using the Solr transport", url, exc + ) + return False + + available = any(getattr(tool, "name", None) == tool_name for tool in tools) + logger.info( + "OKP MCP probe of %r: tool %r %s", + url, + tool_name, + "available" if available else "not advertised", + ) + return available diff --git a/src/pydantic_ai_lightspeed/retrieval/okp_mcp/_provider.py b/src/pydantic_ai_lightspeed/retrieval/okp_mcp/_provider.py new file mode 100644 index 000000000..3a15e3e61 --- /dev/null +++ b/src/pydantic_ai_lightspeed/retrieval/okp_mcp/_provider.py @@ -0,0 +1,338 @@ +"""OKP-over-MCP RAG retriever. + +Fetches OKP RAG context from the RHOKP MCP server and maps the returned +documents to the backend-neutral :class:`~models.common.turn_summary.RAGChunk` +and :class:`~models.common.turn_summary.ReferencedDocument` models. This is the +Pydantic-AI analog of the OGX/Solr ``_fetch_okp_rag`` path; the two are +interchangeable at the ``build_rag_context`` fork and produce the same output +contract. +""" + +from __future__ import annotations + +import traceback +from typing import Any, Optional +from urllib.parse import urljoin + +from pydantic import AnyUrl, ValidationError + +import constants +from configuration import configuration, okp_mcp_endpoint_url +from log import get_logger +from models.common.query import OkpFilter +from models.common.turn_summary import RAGChunk, ReferencedDocument +from pydantic_ai_lightspeed.retrieval.okp_mcp._client import ( + OkpMcpUnavailableError, + call_okp_search, +) + +logger = get_logger(__name__) + + +class OkpMcpRetriever: # pylint: disable=too-many-instance-attributes + """Retrieve OKP RAG context from the RHOKP MCP server. + + Attributes: + url: RHOKP MCP endpoint (streamable HTTP). + tool_name: Name of the MCP search tool to call. + max_chunks: Maximum number of chunks to keep from the response. + offline: When True, build document URLs from ``source_path``; when + False, use ``online_source_url``. + doc_base_url: Base URL used to build offline document URLs. + headers: Optional static request headers (e.g. authorization). + timeout: Optional per-request timeout in seconds. + """ + + def __init__( # pylint: disable=too-many-arguments,too-many-positional-arguments + self, + url: str, + tool_name: str, + max_chunks: int, + offline: bool, + doc_base_url: str, + headers: Optional[dict[str, str]] = None, + timeout: Optional[float] = None, + ) -> None: + """Initialize the retriever with an explicit configuration. + + Prefer :meth:`from_configuration` for the runtime instance; the explicit + constructor exists mainly for testing. + """ + self.url = url + self.tool_name = tool_name + self.max_chunks = max_chunks + self.offline = offline + self.doc_base_url = doc_base_url + self.headers = headers + self.timeout = timeout + + @classmethod + def from_configuration(cls) -> OkpMcpRetriever: + """Build a retriever from the loaded global configuration. + + Reads ``rag.okp``. The MCP endpoint is derived from ``rhokp_url`` (the + RHOKP MCP server is always served at its ``/mcp`` path); the document + base URL falls back to the constant default when ``rhokp_url`` is unset. + Product/version filtering is query-time only, so no launch-time filter + defaults are read. + + Returns: + OkpMcpRetriever: Configured retriever instance. + """ + okp = configuration.okp + doc_base_url = ( + str(okp.rhokp_url) + if okp.rhokp_url is not None + else constants.RH_SERVER_OKP_DEFAULT_URL + ) + return cls( + url=okp_mcp_endpoint_url(), + tool_name=constants.OKP_MCP_DEFAULT_TOOL_NAME, + max_chunks=okp.max_chunks, + offline=okp.offline, + doc_base_url=doc_base_url, + ) + + @staticmethod + def _okp_products_arg( + okp: Optional[OkpFilter], + ) -> Optional[list[dict[str, Any]]]: + """Build the MCP ``products`` filter argument from a query-time filter. + + Translates the transport-neutral + :class:`~models.common.query.OkpFilter` into the RHOKP MCP ``search`` + tool's structured ``products`` argument: a list of + ``{"product": ..., "versions": [...]}`` entries (products OR-combined; + versions within a product OR-combined). The ``versions`` key is omitted + for a product carrying no version restriction. The RHOKP MCP server + turns this structure into the Solr ``fq`` clause, so no per-(product, + version) fan-out is needed on this side. + + Parameters: + okp: Optional query-time OKP filter. + + Returns: + The structured products list, or None when no filter is supplied + (the search then spans all products). + """ + if okp is None or not okp.products: + return None + products: list[dict[str, Any]] = [] + for entry in okp.products: + item: dict[str, Any] = {"product": entry.product} + if entry.versions: + item["versions"] = list(entry.versions) + products.append(item) + return products + + async def fetch( + self, query: str, okp: Optional[OkpFilter] = None + ) -> tuple[list[RAGChunk], list[ReferencedDocument]]: + """Fetch chunks and referenced documents from the RHOKP MCP server. + + Issues a single MCP ``search`` call, passing ``okp`` as the tool's + structured ``products`` filter; the RHOKP MCP server builds the + query-side (Solr ``fq``) filter itself, so a multi-product/multi-version + selection needs no fan-out. Returned documents are deduplicated, sorted + by score, and capped at ``max_chunks``. A transport or tool error causes + the MCP endpoint to be treated as unusable for this request: + :class:`~pydantic_ai_lightspeed.retrieval.okp_mcp._client.OkpMcpUnavailableError` + is raised so the caller can fall back to the Solr transport. + + Parameters: + query: The raw user query string. + okp: Optional query-time OKP filter selecting products/versions. + + Returns: + A tuple of ``(rag_chunks, referenced_documents)``. Both lists are + empty when the server returns no usable documents (a legitimate + zero-result search, distinct from a transport failure). + + Raises: + OkpMcpUnavailableError: When the search failed, signalling that the + caller should fall back to the Solr transport. + """ + rows = min(self.max_chunks, constants.OKP_MCP_MAX_ROWS) + products = self._okp_products_arg(okp) + try: + result = await call_okp_search( + url=self.url, + tool_name=self.tool_name, + query=query, + rows=rows, + headers=self.headers, + timeout=self.timeout, + products=products, + ) + except Exception as exc: # pylint: disable=broad-exception-caught + # Treat the MCP endpoint as unusable for this request so the caller + # can fall back to the Solr transport, rather than silently returning + # an empty result that looks like a legitimate zero-result search. + logger.warning("Failed to query OKP MCP server for chunks: %s", exc) + logger.debug( + "OKP MCP query error details: %s", + "".join(traceback.format_exception(exc)), + ) + raise OkpMcpUnavailableError("OKP MCP search failed for the query") from exc + + docs = self._extract_docs(result) + if not docs: + logger.debug("OKP MCP returned no documents for query") + return [], [] + + docs = self._merge_docs(docs)[: self.max_chunks] + rag_chunks = self._to_rag_chunks(docs) + referenced_documents = self._to_referenced_documents(docs) + logger.debug( + "OKP MCP retrieval: %d chunks, %d documents", + len(rag_chunks), + len(referenced_documents), + ) + return rag_chunks, referenced_documents + + @staticmethod + def _merge_docs(docs: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Merge documents from one or more searches into a ranked, unique list. + + Sorts by descending score (missing scores rank last) and deduplicates + exact repeats — the same chunk of the same document returned by + overlapping searches — while preserving distinct chunks of one document. + + Parameters: + docs: Concatenated document mappings from all issued searches. + + Returns: + Documents sorted by descending score with exact duplicates removed. + """ + + def _score(doc: dict[str, Any]) -> float: + value = doc.get("score") + return float(value) if isinstance(value, (int, float)) else float("-inf") + + docs_sorted = sorted(docs, key=_score, reverse=True) + seen: set[tuple[Any, Any]] = set() + merged: list[dict[str, Any]] = [] + for doc in docs_sorted: + key = (doc.get("doc_id"), doc.get("chunk")) + if key in seen: + continue + seen.add(key) + merged.append(doc) + return merged + + @staticmethod + def _extract_docs(result: dict[str, Any]) -> list[dict[str, Any]]: + """Extract the ``response.docs`` list from a search tool result. + + Parameters: + result: The structured content returned by the MCP search tool. + + Returns: + The list of document mappings, or an empty list when the payload is + missing or malformed. + """ + response = result.get("response") + if not isinstance(response, dict): + return [] + docs = response.get("docs") + if not isinstance(docs, list): + return [] + return [doc for doc in docs if isinstance(doc, dict)] + + def _build_doc_url(self, doc: dict[str, Any]) -> Optional[str]: + """Build a document URL for a single document. + + Uses ``source_path`` (joined onto the OKP base URL) when offline, or the + absolute ``online_source_url`` when online. + + Parameters: + doc: A single document mapping from the MCP search result. + + Returns: + The document URL string, or None when no URL can be built. + """ + if self.offline: + source_path = doc.get("source_path") + if source_path: + return urljoin(self.doc_base_url, source_path) + return None + online_source_url = doc.get("online_source_url") + return online_source_url or None + + def _to_rag_chunks(self, docs: list[dict[str, Any]]) -> list[RAGChunk]: + """Convert MCP documents to ``RAGChunk`` objects. + + Documents without chunk content are skipped. + + Parameters: + docs: Document mappings from the MCP search result. + + Returns: + List of ``RAGChunk`` labelled with the OKP source id. + """ + rag_chunks: list[RAGChunk] = [] + for doc in docs: + content = doc.get("chunk") + if not content: + continue + attributes: dict[str, Any] = {} + doc_url = self._build_doc_url(doc) + if doc_url: + attributes["doc_url"] = doc_url + for key in ("doc_id", "title", "product", "product_version"): + value = doc.get(key) + if value is not None: + attributes[key] = value + if doc.get("doc_id") is not None: + attributes["document_id"] = doc["doc_id"] + + rag_chunks.append( + RAGChunk( + content=content, + source=constants.OKP_RAG_ID, + score=doc.get("score"), + attributes=attributes or None, + ) + ) + return rag_chunks + + def _to_referenced_documents( + self, docs: list[dict[str, Any]] + ) -> list[ReferencedDocument]: + """Extract unique referenced documents from MCP documents. + + Deduplicates by document URL (falling back to ``doc_id``), mirroring the + Solr path so the downstream merge/dedup stays source-agnostic. + + Parameters: + docs: Document mappings from the MCP search result. + + Returns: + List of unique ``ReferencedDocument`` objects. + """ + referenced_documents: list[ReferencedDocument] = [] + seen: set[str] = set() + for doc in docs: + doc_id = doc.get("doc_id") + doc_url = self._build_doc_url(doc) + dedup_key = doc_url or doc_id + if not dedup_key or dedup_key in seen: + continue + seen.add(dedup_key) + + parsed_url: Optional[AnyUrl] = None + if doc_url: + try: + parsed_url = AnyUrl(doc_url) + except ValidationError: + parsed_url = None + + referenced_documents.append( + ReferencedDocument( + doc_title=doc.get("title"), + doc_url=parsed_url, + source=constants.OKP_RAG_ID, + document_id=doc_id, + ) + ) + return referenced_documents diff --git a/src/utils/vector_search.py b/src/utils/vector_search.py index a19ebd063..b3a4006f6 100644 --- a/src/utils/vector_search.py +++ b/src/utils/vector_search.py @@ -1,3 +1,4 @@ +# pylint: disable=too-many-lines """Vector search utilities for query endpoints. This module contains common functionality for performing vector searches @@ -17,11 +18,17 @@ from pydantic import AnyUrl, ValidationError import constants -from configuration import configuration +from configuration import ( + configuration, + mark_okp_mcp_unavailable, + okp_mcp_available, +) from log import get_logger -from models.common.query import SolrVectorSearchRequest +from models.common.query import OkpFilter, SolrVectorSearchRequest from models.common.responses.types import ResponseInput from models.common.turn_summary import RAGChunk, RAGContext, ReferencedDocument +from pydantic_ai_lightspeed.retrieval.okp_mcp import OkpMcpRetriever +from pydantic_ai_lightspeed.retrieval.okp_mcp._client import OkpMcpUnavailableError from utils.otel_tracing import ( SpanAttributes, SpanEvents, @@ -109,15 +116,60 @@ def _get_solr_vector_store_ids() -> list[str]: return vector_store_ids +def _okp_filter_to_structured(okp: Optional[OkpFilter]) -> Optional[dict[str, Any]]: + """Translate a transport-neutral OKP filter into an OGX structured filter. + + Each product selection becomes an exact ``product`` match, AND-combined with + an ``in`` over its versions when versions are given; the selections are then + OR-combined. This is the legacy Solr-path translation and is deleted with the + OGX/Solr transport; the public :class:`OkpFilter` interface is unaffected. + + Parameters: + okp: Optional query-time OKP filter. + + Returns: + An OGX ``{type, key, value}`` filter dict, or None when ``okp`` is None + or selects no products. + """ + if okp is None or not okp.products: + return None + + product_filters: list[dict[str, Any]] = [] + for entry in okp.products: + product_clause: dict[str, Any] = { + "type": "eq", + "key": "product", + "value": entry.product, + } + if entry.versions: + version_clause: dict[str, Any] = { + "type": "in", + "key": "product_version", + "value": list(entry.versions), + } + product_filters.append( + {"type": "and", "filters": [product_clause, version_clause]} + ) + else: + product_filters.append(product_clause) + + if len(product_filters) == 1: + return product_filters[0] + return {"type": "or", "filters": product_filters} + + def _build_query_params( solr: Optional[SolrVectorSearchRequest] = None, max_chunks: Optional[int] = None, + okp: Optional[OkpFilter] = None, ) -> dict[str, Any]: """Build query parameters for Solr vector_io search. Args: solr: Optional structured Solr request (mode and filters from the API). max_chunks: Optional number of chunks to return. If not provided, uses default. + okp: Optional transport-neutral OKP filter, translated to an OGX + structured filter and AND-combined with any structured Solr filter. Returns: Query parameters dict for vector_io.query. @@ -168,6 +220,15 @@ def _build_query_params( else: logger.debug("No solr filters provided") + okp_filter = _okp_filter_to_structured(okp) + if okp_filter is not None: + existing = params.get("filters") + if existing is not None: + params["filters"] = {"type": "and", "filters": [existing, okp_filter]} + else: + params["filters"] = okp_filter + logger.debug("Applied OKP structured filter: %s", params["filters"]) + logger.debug("Final params being sent to vector_io.query: %s", params) return params @@ -579,6 +640,7 @@ async def _fetch_okp_rag( # pylint: disable=too-many-locals client: AsyncOgxClient, query: str, solr: Optional[SolrVectorSearchRequest] = None, + okp: Optional[OkpFilter] = None, ) -> tuple[list[RAGChunk], list[ReferencedDocument]]: """Fetch chunks and documents from Solr RAG source. @@ -586,6 +648,8 @@ async def _fetch_okp_rag( # pylint: disable=too-many-locals client: The AsyncOgxClient to use for the request query: The user's query solr: Structured Solr inline RAG request from the API (optional). + okp: Transport-neutral OKP filter from the API (optional), translated to + an OGX structured filter for the Solr query. Returns: Tuple containing: @@ -609,7 +673,7 @@ async def _fetch_okp_rag( # pylint: disable=too-many-locals if vector_store_ids: # Assuming only one Solr vector store is registered vector_store_id = vector_store_ids[0] - params = _build_query_params(solr, max_chunks=limit) + params = _build_query_params(solr, max_chunks=limit, okp=okp) query_response = await client.vector_io.query( vector_store_id=vector_store_id, @@ -646,12 +710,83 @@ async def _fetch_okp_rag( # pylint: disable=too-many-locals return rag_chunks, referenced_documents -async def build_rag_context( # pylint: disable=too-many-locals,too-many-branches +async def _fetch_okp_rag_mcp( + query: str, + okp: Optional[OkpFilter] = None, +) -> tuple[list[RAGChunk], list[ReferencedDocument]]: + """Fetch chunks and documents from the OKP MCP transport. + + The MCP counterpart of :func:`_fetch_okp_rag`. Unlike the Solr path it does + not need the OGX ``client`` or a ``SolrVectorSearchRequest``: the RHOKP MCP + server encapsulates embeddings and querying. Invoked by :func:`_fetch_okp` + when the RHOKP endpoint has probed as MCP-capable. + + Parameters: + query: The user's query. + okp: Transport-neutral OKP filter from the API (optional). + + Returns: + Tuple containing: + - rag_chunks: RAG chunks from the OKP MCP server. + - referenced_documents: Documents referenced in the MCP results. + + Raises: + OkpMcpUnavailableError: When every MCP search failed, so the caller can + fall back to the Solr transport. + """ + if not configuration.okp_inline_enabled: + logger.info("OKP is disabled for inline RAG, skipping OKP MCP search") + return [], [] + + retriever = OkpMcpRetriever.from_configuration() + return await retriever.fetch(query, okp=okp) + + +async def _fetch_okp( + client: AsyncOgxClient, + query: str, + solr: Optional[SolrVectorSearchRequest] = None, + okp: Optional[OkpFilter] = None, +) -> tuple[list[RAGChunk], list[ReferencedDocument]]: + """Fetch OKP RAG context, preferring the MCP transport with Solr fallback. + + At launch the Solr ``vector_io`` provider is always wired; at query time the + RHOKP MCP transport is attempted whenever the endpoint has probed as + MCP-capable (see :func:`configuration.okp_mcp_available`, which re-probes + periodically so an upgraded RHOKP is adopted without restarting). If MCP is + not available, or a believed-available endpoint hard-fails for this request, + the Solr transport serves it instead. Both honour the same query-time + ``okp`` filter and return the same ``(chunks, documents)`` contract. + + Parameters: + client: OGX client used by the Solr transport. + query: The user's query. + solr: Structured Solr inline RAG request from the API (optional). + okp: Transport-neutral OKP filter from the API (optional). + + Returns: + Tuple of ``(rag_chunks, referenced_documents)`` from whichever transport + served the request. + """ + if await okp_mcp_available(): + try: + return await _fetch_okp_rag_mcp(query, okp) + except OkpMcpUnavailableError: + logger.warning( + "OKP MCP transport failed for this request; " + "falling back to the Solr transport" + ) + mark_okp_mcp_unavailable() + return await _fetch_okp_rag(client, query, solr, okp) + + +async def build_rag_context( # pylint: disable=too-many-locals,too-many-branches,too-many-arguments,too-many-positional-arguments client: AsyncOgxClient, moderation_decision: str, # pylint: disable=unused-argument query: str, vector_store_ids: Optional[list[str]], solr: Optional[SolrVectorSearchRequest] = None, + okp: Optional[OkpFilter] = None, ) -> RAGContext: """Build RAG context by fetching and merging chunks from all enabled sources. @@ -665,6 +800,8 @@ async def build_rag_context( # pylint: disable=too-many-locals,too-many-branche query: The user's query vector_store_ids: The vector store IDs to query solr: Structured Solr inline RAG request from the API (optional). + okp: Transport-neutral OKP filter from the API (optional), applied by + whichever OKP transport is active (RHOKP MCP or the legacy Solr path). Returns: RAGContext containing formatted context text and referenced documents @@ -679,12 +816,17 @@ async def build_rag_context( # pylint: disable=too-many-locals,too-many-branche top_k = configuration.rag.retrieval.inline.max_chunks - # Fetch from each source using per-source limits for the reranking pool + # Fetch from each source using per-source limits for the reranking pool. + # The OKP source has two interchangeable transports selected at query + # time by _fetch_okp: the RHOKP MCP path (preferred when available) and + # the OGX/Solr vector_io path (always-wired fallback). Both return the + # same (chunks, documents) contract so the merge/rerank pipeline is + # unchanged. byok_chunks_task = _fetch_byok_rag(client, query, vector_store_ids) - solr_chunks_task = _fetch_okp_rag(client, query, solr) + okp_chunks_task = _fetch_okp(client, query, solr, okp) (byok_chunks, byok_documents), (solr_chunks, solr_documents) = ( - await asyncio.gather(byok_chunks_task, solr_chunks_task) + await asyncio.gather(byok_chunks_task, okp_chunks_task) ) # Merge chunks diff --git a/tests/unit/models/config/test_rag_configuration.py b/tests/unit/models/config/test_rag_configuration.py index 117986ed6..500fecf52 100644 --- a/tests/unit/models/config/test_rag_configuration.py +++ b/tests/unit/models/config/test_rag_configuration.py @@ -236,6 +236,21 @@ def test_no_unknown_fields_allowed(self) -> None: with pytest.raises(ValidationError, match="Extra inputs are not permitted"): OkpConfiguration(unknown_field="value") # type: ignore[call-arg] + def test_rhokp_url_defaults_to_none(self) -> None: + """Test that rhokp_url is unset by default (endpoint derives a default).""" + config = OkpConfiguration() + assert config.rhokp_url is None + + def test_rhokp_url_accepts_pre_mcp_value(self) -> None: + """The pre-MCP rhokp_url field keeps working unchanged (no mcp block).""" + config = OkpConfiguration(rhokp_url="http://rhokp:8081") # type: ignore[arg-type] + assert str(config.rhokp_url).rstrip("/") == "http://rhokp:8081" + + def test_no_mcp_field(self) -> None: + """The removed nested mcp block is rejected as an unknown field.""" + with pytest.raises(ValidationError, match="Extra inputs are not permitted"): + OkpConfiguration(mcp={"enabled": True}) # type: ignore[call-arg] + class TestOldFormatRejected: """Tests that old-style RAG config fields are rejected.""" diff --git a/tests/unit/models/requests/test_query_request.py b/tests/unit/models/requests/test_query_request.py index 869b16044..6fc6d2bd1 100644 --- a/tests/unit/models/requests/test_query_request.py +++ b/tests/unit/models/requests/test_query_request.py @@ -1,9 +1,14 @@ """Unit tests for QueryRequest model.""" +# pylint: disable=no-member +# Pydantic Optional model fields confuse pylint's attribute inference after +# a narrowing ``assert ... is not None``. + import pytest +from pydantic import ValidationError from models.api.requests import QueryRequest -from models.common.query import Attachment, SolrVectorSearchRequest +from models.common.query import Attachment, OkpFilter, SolrVectorSearchRequest class TestQueryRequest: @@ -147,3 +152,50 @@ def test_solr_structured_mode_and_filters(self) -> None: solr_request = SolrVectorSearchRequest.model_validate(qr.solr) assert solr_request.mode == "hybrid" assert solr_request.filters == {"fq": ["x:y"]} + + def test_okp_filter_parsed(self) -> None: + """The transport-neutral ``okp`` filter is parsed into typed objects.""" + qr = QueryRequest( + query="q", + okp={ + "products": [ + { + "product": "openshift_container_platform", + "versions": ["4.16", "4.17"], + } + ] + }, + ) # pyright: ignore[reportCallIssue] + assert qr.okp is not None + assert qr.okp.products[0].product == "openshift_container_platform" + assert qr.okp.products[0].versions == ["4.16", "4.17"] + + def test_okp_defaults_to_none(self) -> None: + """The ``okp`` field is optional and defaults to None.""" + qr = QueryRequest(query="q") + assert qr.okp is None + + +class TestOkpFilter: + """Tests for the OkpFilter / OkpProductFilter request models.""" + + def test_product_is_required(self) -> None: + """A product selection without a product identifier is rejected.""" + with pytest.raises(ValidationError): + OkpFilter.model_validate({"products": [{"versions": ["4.16"]}]}) + + def test_versions_optional(self) -> None: + """Versions may be omitted, matching the product regardless of version.""" + okp = OkpFilter.model_validate( + {"products": [{"product": "openshift_container_platform"}]} + ) + assert okp.products[0].versions is None + + def test_products_defaults_to_empty_list(self) -> None: + """An OkpFilter with no products is an empty (no-op) filter.""" + assert OkpFilter().products == [] + + def test_unknown_fields_rejected(self) -> None: + """Unknown fields are rejected on both nested models.""" + with pytest.raises(ValidationError): + OkpFilter.model_validate({"products": [{"product": "p", "bogus": 1}]}) diff --git a/tests/unit/pydantic_ai_lightspeed/retrieval/__init__.py b/tests/unit/pydantic_ai_lightspeed/retrieval/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/pydantic_ai_lightspeed/retrieval/okp_mcp/__init__.py b/tests/unit/pydantic_ai_lightspeed/retrieval/okp_mcp/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/unit/pydantic_ai_lightspeed/retrieval/okp_mcp/test_client.py b/tests/unit/pydantic_ai_lightspeed/retrieval/okp_mcp/test_client.py new file mode 100644 index 000000000..1eea7c5bf --- /dev/null +++ b/tests/unit/pydantic_ai_lightspeed/retrieval/okp_mcp/test_client.py @@ -0,0 +1,197 @@ +"""Unit tests for the OKP MCP client wrapper.""" + +from typing import Any + +import pytest +from pytest_mock import MockerFixture + +from pydantic_ai_lightspeed.retrieval.okp_mcp import _client + + +@pytest.mark.asyncio +async def test_call_okp_search_returns_structured_mapping( + mocker: MockerFixture, +) -> None: + """A mapping result from the tool is returned as a plain dict.""" + payload = {"response": {"numFound": 1, "docs": [{"chunk": "x"}]}} + toolset = mocker.Mock() + toolset.direct_call_tool = mocker.AsyncMock(return_value=payload) + toolset_cls = mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + result = await _client.call_okp_search( + url="http://okp/mcp", tool_name="search", query="q", rows=5 + ) + + assert result == payload + # URL is the positional client arg; error behavior is strict. + args, kwargs = toolset_cls.call_args + assert args[0] == "http://okp/mcp" + assert kwargs["tool_error_behavior"] == "error" + toolset.direct_call_tool.assert_awaited_once_with( + "search", {"query": "q", "rows": 5} + ) + + +@pytest.mark.asyncio +async def test_call_okp_search_forwards_headers_and_timeout( + mocker: MockerFixture, +) -> None: + """Headers and timeout are forwarded to the toolset when provided.""" + toolset = mocker.Mock() + toolset.direct_call_tool = mocker.AsyncMock(return_value={}) + toolset_cls = mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + await _client.call_okp_search( + url="http://okp/mcp", + tool_name="search", + query="q", + rows=3, + headers={"Authorization": "Bearer t"}, + timeout=12.0, + ) + + _, kwargs = toolset_cls.call_args + assert kwargs["headers"] == {"Authorization": "Bearer t"} + assert kwargs["init_timeout"] == 12.0 + assert kwargs["read_timeout"] == 12.0 + + +@pytest.mark.asyncio +async def test_call_okp_search_omits_products_when_none( + mocker: MockerFixture, +) -> None: + """The products filter is absent from the tool args when not supplied.""" + toolset = mocker.Mock() + toolset.direct_call_tool = mocker.AsyncMock(return_value={}) + mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + await _client.call_okp_search( + url="http://okp/mcp", tool_name="search", query="q", rows=5 + ) + + toolset.direct_call_tool.assert_awaited_once_with( + "search", {"query": "q", "rows": 5} + ) + + +@pytest.mark.asyncio +async def test_call_okp_search_omits_products_when_empty( + mocker: MockerFixture, +) -> None: + """An empty products list is treated as no filter and omitted.""" + toolset = mocker.Mock() + toolset.direct_call_tool = mocker.AsyncMock(return_value={}) + mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + await _client.call_okp_search( + url="http://okp/mcp", tool_name="search", query="q", rows=5, products=[] + ) + + toolset.direct_call_tool.assert_awaited_once_with( + "search", {"query": "q", "rows": 5} + ) + + +@pytest.mark.asyncio +async def test_call_okp_search_forwards_structured_products( + mocker: MockerFixture, +) -> None: + """A structured products filter is forwarded verbatim in the tool args.""" + toolset = mocker.Mock() + toolset.direct_call_tool = mocker.AsyncMock(return_value={}) + mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + products = [ + {"product": "openshift_container_platform", "versions": ["4.19", "4.20"]}, + {"product": "red_hat_enterprise_linux"}, + ] + await _client.call_okp_search( + url="http://okp/mcp", + tool_name="search", + query="q", + rows=5, + products=products, + ) + + toolset.direct_call_tool.assert_awaited_once_with( + "search", + {"query": "q", "rows": 5, "products": products}, + ) + + +@pytest.mark.asyncio +async def test_call_okp_search_non_mapping_result_returns_empty( + mocker: MockerFixture, +) -> None: + """A non-mapping tool result is ignored and an empty dict is returned.""" + toolset = mocker.Mock() + toolset.direct_call_tool = mocker.AsyncMock(return_value="just text") + mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + result: dict[str, Any] = await _client.call_okp_search( + url="http://okp/mcp", tool_name="search", query="q", rows=5 + ) + + assert result == {} + + +@pytest.mark.asyncio +async def test_probe_okp_mcp_true_when_tool_advertised( + mocker: MockerFixture, +) -> None: + """The probe returns True when the endpoint advertises the search tool.""" + tool = mocker.Mock() + tool.name = "search" + toolset = mocker.Mock() + toolset.list_tools = mocker.AsyncMock(return_value=[tool]) + mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + assert await _client.probe_okp_mcp("http://okp/mcp", "search") is True + + +@pytest.mark.asyncio +async def test_probe_okp_mcp_false_when_tool_absent( + mocker: MockerFixture, +) -> None: + """The probe returns False when the search tool is not advertised.""" + tool = mocker.Mock() + tool.name = "other" + toolset = mocker.Mock() + toolset.list_tools = mocker.AsyncMock(return_value=[tool]) + mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + assert await _client.probe_okp_mcp("http://okp/mcp", "search") is False + + +@pytest.mark.asyncio +async def test_probe_okp_mcp_false_on_transport_error( + mocker: MockerFixture, +) -> None: + """The probe never raises: a transport error degrades to False.""" + toolset = mocker.Mock() + toolset.list_tools = mocker.AsyncMock(side_effect=RuntimeError("no route")) + mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + assert await _client.probe_okp_mcp("http://okp/mcp", "search") is False + + +@pytest.mark.asyncio +async def test_probe_okp_mcp_forwards_headers_and_timeout( + mocker: MockerFixture, +) -> None: + """Headers and timeout are forwarded to the toolset when provided.""" + toolset = mocker.Mock() + toolset.list_tools = mocker.AsyncMock(return_value=[]) + toolset_cls = mocker.patch.object(_client, "MCPToolset", return_value=toolset) + + await _client.probe_okp_mcp( + "http://okp/mcp", + "search", + headers={"Authorization": "Bearer t"}, + timeout=5.0, + ) + + _, kwargs = toolset_cls.call_args + assert kwargs["headers"] == {"Authorization": "Bearer t"} + assert kwargs["init_timeout"] == 5.0 + assert kwargs["read_timeout"] == 5.0 diff --git a/tests/unit/pydantic_ai_lightspeed/retrieval/okp_mcp/test_provider.py b/tests/unit/pydantic_ai_lightspeed/retrieval/okp_mcp/test_provider.py new file mode 100644 index 000000000..3b3d09f04 --- /dev/null +++ b/tests/unit/pydantic_ai_lightspeed/retrieval/okp_mcp/test_provider.py @@ -0,0 +1,313 @@ +"""Unit tests for the OKP MCP retriever.""" + +from typing import Any +from urllib.parse import urljoin + +import pytest +from pydantic import AnyUrl +from pytest_mock import MockerFixture + +import constants +from models.common.query import OkpFilter +from pydantic_ai_lightspeed.retrieval.okp_mcp import _provider +from pydantic_ai_lightspeed.retrieval.okp_mcp._client import OkpMcpUnavailableError +from pydantic_ai_lightspeed.retrieval.okp_mcp._provider import OkpMcpRetriever + +SAMPLE_RESULT: dict[str, Any] = { + "response": { + "numFound": 2, + "docs": [ + { + "chunk": "content A", + "score": 74.0, + "title": "Title A", + "doc_id": "doc-a", + "product": ["red_hat_enterprise_linux"], + "product_version": "9", + "online_source_url": "https://docs.redhat.com/a", + "source_path": "/en/a", + }, + { + "chunk": "content B", + "score": 73.0, + "title": "Title B", + "doc_id": "doc-b", + "online_source_url": "https://docs.redhat.com/b", + "source_path": "/en/b", + }, + ], + } +} + + +def _retriever(offline: bool = False, max_chunks: int = 5) -> OkpMcpRetriever: + """Build a retriever with an explicit, test-friendly configuration.""" + return OkpMcpRetriever( + url="http://okp:8080/mcp", + tool_name="search", + max_chunks=max_chunks, + offline=offline, + doc_base_url="http://okp:8081", + ) + + +@pytest.mark.asyncio +async def test_fetch_maps_online_urls(mocker: MockerFixture) -> None: + """Online mode uses online_source_url and maps chunks + documents.""" + mocker.patch.object( + _provider, "call_okp_search", mocker.AsyncMock(return_value=SAMPLE_RESULT) + ) + + chunks, documents = await _retriever(offline=False).fetch("q") + + assert [c.content for c in chunks] == ["content A", "content B"] + assert all(c.source == constants.OKP_RAG_ID for c in chunks) + assert chunks[0].score == 74.0 + assert chunks[0].attributes["doc_url"] == "https://docs.redhat.com/a" + assert chunks[0].attributes["document_id"] == "doc-a" + assert chunks[0].attributes["product"] == ["red_hat_enterprise_linux"] + + assert [str(d.doc_url) for d in documents] == [ + "https://docs.redhat.com/a", + "https://docs.redhat.com/b", + ] + assert documents[0].doc_title == "Title A" + assert documents[0].source == constants.OKP_RAG_ID + + +@pytest.mark.asyncio +async def test_fetch_maps_offline_urls(mocker: MockerFixture) -> None: + """Offline mode joins source_path onto the document base URL.""" + mocker.patch.object( + _provider, "call_okp_search", mocker.AsyncMock(return_value=SAMPLE_RESULT) + ) + + chunks, documents = await _retriever(offline=True).fetch("q") + + assert chunks[0].attributes["doc_url"] == "http://okp:8081/en/a" + assert documents[0].doc_url == AnyUrl("http://okp:8081/en/a") + + +@pytest.mark.asyncio +async def test_fetch_caps_at_max_chunks(mocker: MockerFixture) -> None: + """Only max_chunks documents are kept.""" + mocker.patch.object( + _provider, "call_okp_search", mocker.AsyncMock(return_value=SAMPLE_RESULT) + ) + + chunks, _ = await _retriever(max_chunks=1).fetch("q") + + assert len(chunks) == 1 + assert chunks[0].content == "content A" + + +@pytest.mark.asyncio +async def test_fetch_requests_clamped_rows(mocker: MockerFixture) -> None: + """rows requested from the server never exceed OKP_MCP_MAX_ROWS.""" + call = mocker.AsyncMock(return_value={"response": {"docs": []}}) + mocker.patch.object(_provider, "call_okp_search", call) + + await _retriever(max_chunks=100).fetch("q") + + assert call.await_args.kwargs["rows"] == constants.OKP_MCP_MAX_ROWS + + +@pytest.mark.asyncio +async def test_fetch_skips_chunks_without_content(mocker: MockerFixture) -> None: + """Documents lacking chunk text produce no RAGChunk.""" + result = {"response": {"docs": [{"doc_id": "d1", "title": "t"}, {"chunk": "keep"}]}} + mocker.patch.object( + _provider, "call_okp_search", mocker.AsyncMock(return_value=result) + ) + + chunks, _ = await _retriever().fetch("q") + + assert [c.content for c in chunks] == ["keep"] + + +@pytest.mark.asyncio +async def test_fetch_dedups_documents(mocker: MockerFixture) -> None: + """Documents with the same URL are deduplicated.""" + result = { + "response": { + "docs": [ + {"chunk": "a", "doc_id": "d", "online_source_url": "https://x/1"}, + {"chunk": "b", "doc_id": "d", "online_source_url": "https://x/1"}, + ] + } + } + mocker.patch.object( + _provider, "call_okp_search", mocker.AsyncMock(return_value=result) + ) + + chunks, documents = await _retriever().fetch("q") + + assert len(chunks) == 2 + assert len(documents) == 1 + + +@pytest.mark.asyncio +async def test_fetch_raises_when_search_fails(mocker: MockerFixture) -> None: + """When the search fails, OkpMcpUnavailableError signals Solr fallback.""" + mocker.patch.object( + _provider, + "call_okp_search", + mocker.AsyncMock(side_effect=RuntimeError("boom")), + ) + + with pytest.raises(OkpMcpUnavailableError): + await _retriever().fetch("q") + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "payload", + [ + {}, + {"response": None}, + {"response": {"docs": None}}, + {"response": {}}, + ], +) +async def test_fetch_handles_malformed_payload( + mocker: MockerFixture, payload: dict[str, Any] +) -> None: + """Missing/malformed response shapes yield empty results.""" + mocker.patch.object( + _provider, "call_okp_search", mocker.AsyncMock(return_value=payload) + ) + + chunks, documents = await _retriever().fetch("q") + + assert chunks == [] + assert documents == [] + + +@pytest.mark.asyncio +async def test_fetch_defaults_products_to_none(mocker: MockerFixture) -> None: + """Without a configured filter, None is forwarded (client then omits it).""" + call = mocker.AsyncMock(return_value={"response": {"docs": []}}) + mocker.patch.object(_provider, "call_okp_search", call) + + await _retriever().fetch("q") + + assert call.await_count == 1 + assert call.await_args.kwargs["products"] is None + + +@pytest.mark.asyncio +async def test_fetch_passes_structured_products_in_one_call( + mocker: MockerFixture, +) -> None: + """A multi-version query-time filter is sent as one structured search call.""" + call = mocker.AsyncMock(return_value={"response": {"docs": []}}) + mocker.patch.object(_provider, "call_okp_search", call) + + okp = OkpFilter.model_validate( + { + "products": [ + { + "product": "openshift_container_platform", + "versions": ["4.16", "4.17"], + }, + {"product": "red_hat_enterprise_linux"}, + ] + } + ) + await _retriever().fetch("q", okp=okp) + + assert call.await_count == 1 + assert call.await_args.kwargs["products"] == [ + { + "product": "openshift_container_platform", + "versions": ["4.16", "4.17"], + }, + {"product": "red_hat_enterprise_linux"}, + ] + + +@pytest.mark.asyncio +async def test_fetch_okp_filter_selects_products(mocker: MockerFixture) -> None: + """A query-time filter drives the structured products argument.""" + call = mocker.AsyncMock(return_value={"response": {"docs": []}}) + mocker.patch.object(_provider, "call_okp_search", call) + + okp = OkpFilter.model_validate( + {"products": [{"product": "red_hat_enterprise_linux"}]} + ) + await _retriever().fetch("q", okp=okp) + + assert call.await_count == 1 + assert call.await_args.kwargs["products"] == [ + {"product": "red_hat_enterprise_linux"} + ] + + +@pytest.mark.asyncio +async def test_fetch_merges_and_dedups_response_docs(mocker: MockerFixture) -> None: + """Docs in a single response are sorted by score and deduplicated.""" + result = { + "response": { + "docs": [ + {"chunk": "low", "doc_id": "e", "score": 10.0}, + {"chunk": "shared", "doc_id": "d", "score": 60.0}, + {"chunk": "high", "doc_id": "f", "score": 90.0}, + {"chunk": "shared", "doc_id": "d", "score": 60.0}, + ] + } + } + mocker.patch.object( + _provider, "call_okp_search", mocker.AsyncMock(return_value=result) + ) + + chunks, _ = await _retriever(max_chunks=5).fetch("q") + + # The duplicate "shared" chunk is removed; results are score-sorted. + assert [c.content for c in chunks] == ["high", "shared", "low"] + + +def test_from_configuration_uses_defaults(mocker: MockerFixture) -> None: + """from_configuration falls back to constant defaults when rhokp_url is unset.""" + okp = mocker.Mock() + okp.rhokp_url = None + okp.offline = True + okp.max_chunks = 7 + config_mock = mocker.Mock() + config_mock.okp = okp + mocker.patch.object(_provider, "configuration", config_mock) + mocker.patch.object( + _provider, + "okp_mcp_endpoint_url", + return_value=urljoin(constants.RH_SERVER_OKP_DEFAULT_URL, "/mcp"), + ) + + retriever = OkpMcpRetriever.from_configuration() + + assert retriever.url == urljoin(constants.RH_SERVER_OKP_DEFAULT_URL, "/mcp") + assert retriever.doc_base_url == constants.RH_SERVER_OKP_DEFAULT_URL + assert retriever.tool_name == constants.OKP_MCP_DEFAULT_TOOL_NAME + assert retriever.max_chunks == 7 + assert retriever.offline is True + assert retriever.headers is None + assert retriever.timeout is None + + +def test_from_configuration_derives_endpoint_from_rhokp_url( + mocker: MockerFixture, +) -> None: + """from_configuration derives the endpoint and doc base URL from rhokp_url.""" + okp = mocker.Mock() + okp.rhokp_url = "http://rhokp:9000" + okp.offline = False + okp.max_chunks = 5 + config_mock = mocker.Mock() + config_mock.okp = okp + mocker.patch.object(_provider, "configuration", config_mock) + mocker.patch.object( + _provider, "okp_mcp_endpoint_url", return_value="http://rhokp:9000/mcp" + ) + + retriever = OkpMcpRetriever.from_configuration() + + assert retriever.url == "http://rhokp:9000/mcp" + assert retriever.doc_base_url == "http://rhokp:9000" diff --git a/tests/unit/test_configuration.py b/tests/unit/test_configuration.py index 6df3a10f4..26ed45d9a 100644 --- a/tests/unit/test_configuration.py +++ b/tests/unit/test_configuration.py @@ -5,19 +5,30 @@ from collections.abc import Generator from pathlib import Path from typing import Any +from urllib.parse import urljoin import pytest from pydantic import ValidationError +from pytest_mock import MockerFixture +import configuration as configuration_module import constants from cache.in_memory_cache import InMemoryCache from cache.sqlite_cache import SQLiteCache from configuration import ( AppConfig, LogicError, + mark_okp_mcp_unavailable, + okp_mcp_available, + okp_mcp_endpoint_url, replace_env_vars_preserving_native_override, + reset_okp_mcp_probe, +) +from models.config import ( + CustomProfile, + ModelContextProtocolServer, + OkpConfiguration, ) -from models.config import CustomProfile, ModelContextProtocolServer from utils.checks import InvalidConfigurationError @@ -4245,3 +4256,103 @@ def test_replace_env_vars_without_native_override_resolves_all( config_dict = {"inference": {"default_model": "${env.LCORE_TEST_MODEL}"}} resolved = replace_env_vars_preserving_native_override(config_dict) assert resolved["inference"]["default_model"] == "gpt-4o-mini" + + +@pytest.fixture(name="_reset_okp_probe") +def _reset_okp_probe_fixture() -> Generator: + """Reset the cached OKP MCP probe result around each transport test.""" + reset_okp_mcp_probe() + yield + reset_okp_mcp_probe() + + +def _patch_okp(monkeypatch: pytest.MonkeyPatch, okp: OkpConfiguration) -> None: + """Point the configuration singleton's ``okp`` property at ``okp``.""" + monkeypatch.setattr( + type(configuration_module.configuration), + "okp", + property(lambda self: okp), + ) + + +def test_okp_mcp_endpoint_url_derived_from_rhokp_url( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The MCP endpoint is the ``/mcp`` path of the configured rhokp_url.""" + _patch_okp(monkeypatch, OkpConfiguration(rhokp_url="http://rhokp.example:8081")) + assert okp_mcp_endpoint_url() == "http://rhokp.example:8081/mcp" + + +def test_okp_mcp_endpoint_url_defaults_when_unset( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """When rhokp_url is unset, the constant default base is used.""" + _patch_okp(monkeypatch, OkpConfiguration()) + assert okp_mcp_endpoint_url() == urljoin( + constants.RH_SERVER_OKP_DEFAULT_URL, "/mcp" + ) + + +@pytest.mark.asyncio +async def test_okp_mcp_available_true_when_probe_succeeds( + monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture, _reset_okp_probe: None +) -> None: + """A successful probe reports MCP available and is cached (probed once).""" + _patch_okp(monkeypatch, OkpConfiguration(rhokp_url="http://rhokp.example:8081")) + probe = mocker.AsyncMock(return_value=True) + monkeypatch.setattr( + "pydantic_ai_lightspeed.retrieval.okp_mcp._client.probe_okp_mcp", probe + ) + + assert await okp_mcp_available() is True + assert await okp_mcp_available() is True # cached + probe.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_okp_mcp_available_false_when_probe_fails( + monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture, _reset_okp_probe: None +) -> None: + """A failing probe reports MCP unavailable (Solr transport used).""" + _patch_okp(monkeypatch, OkpConfiguration(rhokp_url="http://rhokp.example:8081")) + monkeypatch.setattr( + "pydantic_ai_lightspeed.retrieval.okp_mcp._client.probe_okp_mcp", + mocker.AsyncMock(return_value=False), + ) + + assert await okp_mcp_available() is False + + +@pytest.mark.asyncio +async def test_okp_mcp_reprobes_after_ttl( + monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture, _reset_okp_probe: None +) -> None: + """After the TTL elapses the endpoint is re-probed (fail forward).""" + _patch_okp(monkeypatch, OkpConfiguration(rhokp_url="http://rhokp.example:8081")) + probe = mocker.AsyncMock(side_effect=[False, True]) + monkeypatch.setattr( + "pydantic_ai_lightspeed.retrieval.okp_mcp._client.probe_okp_mcp", probe + ) + now = 0.0 + monkeypatch.setattr(configuration_module.time, "monotonic", lambda: now) + + assert await okp_mcp_available() is False # first probe + now = constants.OKP_MCP_PROBE_TTL_SECONDS + 1.0 + assert await okp_mcp_available() is True # re-probed past TTL + assert probe.await_count == 2 + + +@pytest.mark.asyncio +async def test_mark_okp_mcp_unavailable_forces_solr( + monkeypatch: pytest.MonkeyPatch, mocker: MockerFixture, _reset_okp_probe: None +) -> None: + """Marking unavailable pins Solr within the TTL without re-probing.""" + _patch_okp(monkeypatch, OkpConfiguration(rhokp_url="http://rhokp.example:8081")) + probe = mocker.AsyncMock(return_value=True) + monkeypatch.setattr( + "pydantic_ai_lightspeed.retrieval.okp_mcp._client.probe_okp_mcp", probe + ) + + mark_okp_mcp_unavailable() + assert await okp_mcp_available() is False + probe.assert_not_awaited() diff --git a/tests/unit/utils/test_vector_search.py b/tests/unit/utils/test_vector_search.py index a42388763..4b9f70a80 100644 --- a/tests/unit/utils/test_vector_search.py +++ b/tests/unit/utils/test_vector_search.py @@ -13,8 +13,9 @@ import constants from configuration import AppConfig -from models.common.query import SolrVectorSearchRequest +from models.common.query import OkpFilter, SolrVectorSearchRequest from models.common.turn_summary import RAGChunk, ReferencedDocument +from pydantic_ai_lightspeed.retrieval.okp_mcp._client import OkpMcpUnavailableError from utils.otel_tracing import SpanAttributes, SpanEvents from utils.reranker import ( _get_cross_encoder, @@ -28,16 +29,33 @@ _extract_byok_rag_chunks, _extract_solr_document_metadata, _fetch_byok_rag, + _fetch_okp, _fetch_okp_rag, + _fetch_okp_rag_mcp, _format_rag_context, _get_okp_base_url, _get_solr_vector_store_ids, _is_solr_enabled, + _okp_filter_to_structured, _query_store_for_byok_rag, build_rag_context, ) +@pytest.fixture(autouse=True) +def _force_okp_mcp_unavailable(mocker: MockerFixture) -> None: + """Force the OKP MCP probe off so unit tests never make a live probe call. + + ``_fetch_okp`` probes the RHOKP endpoint at query time; without this the + Solr-transport tests would hit the network and be non-deterministic. Tests + exercising the MCP transport re-patch ``okp_mcp_available`` after this runs. + """ + mocker.patch( + "utils.vector_search.okp_mcp_available", + new=mocker.AsyncMock(return_value=False), + ) + + def _vector_io_query_stub_like_backend( chunk_score_pairs: list[tuple[Any, float]], mocker: MockerFixture ) -> Callable[..., Awaitable[Any]]: @@ -283,6 +301,113 @@ def test_lexical_config_translated_to_keyword(self, mocker: MockerFixture) -> No assert params["mode"] == "keyword" +class TestOkpFilterToStructured: + """Tests for _okp_filter_to_structured translation.""" + + def test_none_returns_none(self) -> None: + """A None filter yields no structured filter.""" + assert _okp_filter_to_structured(None) is None + + def test_empty_products_returns_none(self) -> None: + """An empty product list yields no structured filter.""" + assert _okp_filter_to_structured(OkpFilter(products=[])) is None + + def test_single_product_no_versions(self) -> None: + """A single product without versions becomes a bare eq clause.""" + okp = OkpFilter.model_validate( + {"products": [{"product": "openshift_container_platform"}]} + ) + result = _okp_filter_to_structured(okp) + assert result == { + "type": "eq", + "key": "product", + "value": "openshift_container_platform", + } + + def test_single_product_with_versions(self) -> None: + """Versions become an AND of product eq and version in.""" + okp = OkpFilter.model_validate( + { + "products": [ + { + "product": "openshift_container_platform", + "versions": ["4.16", "4.17"], + } + ] + } + ) + result = _okp_filter_to_structured(okp) + assert result == { + "type": "and", + "filters": [ + { + "type": "eq", + "key": "product", + "value": "openshift_container_platform", + }, + {"type": "in", "key": "product_version", "value": ["4.16", "4.17"]}, + ], + } + + def test_multiple_products_are_ored(self) -> None: + """Multiple product selections are OR-combined.""" + okp = OkpFilter.model_validate( + { + "products": [ + {"product": "openshift_container_platform", "versions": ["4.16"]}, + {"product": "red_hat_enterprise_linux"}, + ] + } + ) + result = _okp_filter_to_structured(okp) + assert result is not None + assert result["type"] == "or" + assert len(result["filters"]) == 2 + assert result["filters"][1] == { + "type": "eq", + "key": "product", + "value": "red_hat_enterprise_linux", + } + + +class TestBuildQueryParamsOkp: + """Tests for _build_query_params OKP filter handling.""" + + def test_okp_only_sets_filters(self) -> None: + """An OKP filter alone populates params['filters'].""" + okp = OkpFilter.model_validate( + {"products": [{"product": "openshift_container_platform"}]} + ) + params = _build_query_params(okp=okp) + + assert params["filters"] == { + "type": "eq", + "key": "product", + "value": "openshift_container_platform", + } + + def test_okp_and_structured_solr_are_anded(self) -> None: + """OKP filter is AND-combined with an existing structured solr filter.""" + solr = SolrVectorSearchRequest.model_validate( + {"filters": {"filters": {"type": "eq", "key": "lang", "value": "en"}}} + ) + okp = OkpFilter.model_validate( + {"products": [{"product": "openshift_container_platform"}]} + ) + params = _build_query_params(solr=solr, okp=okp) + + assert params["filters"]["type"] == "and" + assert params["filters"]["filters"] == [ + {"type": "eq", "key": "lang", "value": "en"}, + {"type": "eq", "key": "product", "value": "openshift_container_platform"}, + ] + + def test_no_okp_leaves_filters_absent(self) -> None: + """Without an OKP filter (or solr filters), no filters key is set.""" + params = _build_query_params() + assert "filters" not in params + + class TestExtractByokRagChunks: """Tests for _extract_byok_rag_chunks function.""" @@ -1818,3 +1943,186 @@ async def test_passed_with_chunks_sets_sources_and_chunk_count( completed_attrs = completed.attributes assert completed_attrs is not None assert completed_attrs["rag.chunks.count"] == 1 + + +class TestFetchOkpRagMcp: + """Tests for the _fetch_okp_rag_mcp OKP MCP transport helper.""" + + @pytest.mark.asyncio + async def test_returns_empty_when_okp_inline_disabled( + self, mocker: MockerFixture + ) -> None: + """When OKP is not an inline source, no MCP call is made.""" + config_mock = mocker.Mock(spec=AppConfig) + config_mock.okp_inline_enabled = False + mocker.patch("utils.vector_search.configuration", config_mock) + retriever_cls = mocker.patch("utils.vector_search.OkpMcpRetriever") + + chunks, documents = await _fetch_okp_rag_mcp("test query") + + assert chunks == [] + assert documents == [] + retriever_cls.from_configuration.assert_not_called() + + @pytest.mark.asyncio + async def test_delegates_to_retriever_when_enabled( + self, mocker: MockerFixture + ) -> None: + """When OKP inline is enabled, the retriever fetches for the query.""" + config_mock = mocker.Mock(spec=AppConfig) + config_mock.okp_inline_enabled = True + mocker.patch("utils.vector_search.configuration", config_mock) + + expected = ( + [RAGChunk(content="c", source=constants.OKP_RAG_ID, score=1.0)], + [ReferencedDocument(doc_title="t", source=constants.OKP_RAG_ID)], + ) + retriever = mocker.Mock() + retriever.fetch = mocker.AsyncMock(return_value=expected) + retriever_cls = mocker.patch("utils.vector_search.OkpMcpRetriever") + retriever_cls.from_configuration.return_value = retriever + + result = await _fetch_okp_rag_mcp("test query") + + assert result == expected + retriever.fetch.assert_awaited_once_with("test query", okp=None) + + +class TestFetchOkpTransportSelection: + """Tests for the OKP MCP/Solr transport selection in _fetch_okp.""" + + @pytest.mark.asyncio + async def test_uses_mcp_transport_when_available( + self, mocker: MockerFixture + ) -> None: + """When MCP has probed available, the MCP path is used and Solr skipped.""" + mocker.patch( + "utils.vector_search.okp_mcp_available", + new=mocker.AsyncMock(return_value=True), + ) + mcp_fetch = mocker.patch( + "utils.vector_search._fetch_okp_rag_mcp", + mocker.AsyncMock( + return_value=( + [RAGChunk(content="mcp", source=constants.OKP_RAG_ID, score=1.0)], + [], + ) + ), + ) + solr_fetch = mocker.patch( + "utils.vector_search._fetch_okp_rag", + mocker.AsyncMock(return_value=([], [])), + ) + + client_mock = mocker.AsyncMock() + chunks, _ = await _fetch_okp(client_mock, "test query") + + mcp_fetch.assert_awaited_once() + solr_fetch.assert_not_called() + assert any(c.content == "mcp" for c in chunks) + + @pytest.mark.asyncio + async def test_uses_solr_transport_when_unavailable( + self, mocker: MockerFixture + ) -> None: + """When MCP is unavailable, the Solr path is used and MCP path is skipped.""" + mocker.patch( + "utils.vector_search.okp_mcp_available", + new=mocker.AsyncMock(return_value=False), + ) + mcp_fetch = mocker.patch( + "utils.vector_search._fetch_okp_rag_mcp", + mocker.AsyncMock(return_value=([], [])), + ) + solr_fetch = mocker.patch( + "utils.vector_search._fetch_okp_rag", + mocker.AsyncMock( + return_value=( + [RAGChunk(content="solr", source=constants.OKP_RAG_ID, score=1.0)], + [], + ) + ), + ) + + client_mock = mocker.AsyncMock() + chunks, _ = await _fetch_okp(client_mock, "test query") + + solr_fetch.assert_awaited_once() + mcp_fetch.assert_not_called() + assert any(c.content == "solr" for c in chunks) + + @pytest.mark.asyncio + async def test_falls_back_to_solr_when_mcp_hard_fails( + self, mocker: MockerFixture + ) -> None: + """A believed-available endpoint that hard-fails falls forward to Solr.""" + mocker.patch( + "utils.vector_search.okp_mcp_available", + new=mocker.AsyncMock(return_value=True), + ) + mocker.patch( + "utils.vector_search._fetch_okp_rag_mcp", + mocker.AsyncMock(side_effect=OkpMcpUnavailableError("boom")), + ) + mark = mocker.patch("utils.vector_search.mark_okp_mcp_unavailable") + solr_fetch = mocker.patch( + "utils.vector_search._fetch_okp_rag", + mocker.AsyncMock( + return_value=( + [RAGChunk(content="solr", source=constants.OKP_RAG_ID, score=1.0)], + [], + ) + ), + ) + + client_mock = mocker.AsyncMock() + chunks, _ = await _fetch_okp(client_mock, "test query") + + mark.assert_called_once() + solr_fetch.assert_awaited_once() + assert any(c.content == "solr" for c in chunks) + + @pytest.mark.asyncio + async def test_forwards_okp_filter_to_mcp_transport( + self, mocker: MockerFixture + ) -> None: + """The request-level okp filter is forwarded to the MCP transport.""" + mocker.patch( + "utils.vector_search.okp_mcp_available", + new=mocker.AsyncMock(return_value=True), + ) + mcp_fetch = mocker.patch( + "utils.vector_search._fetch_okp_rag_mcp", + mocker.AsyncMock(return_value=([], [])), + ) + + okp = OkpFilter.model_validate( + {"products": [{"product": "red_hat_enterprise_linux"}]} + ) + client_mock = mocker.AsyncMock() + await _fetch_okp(client_mock, "q", okp=okp) + + mcp_fetch.assert_awaited_once_with("q", okp) + + @pytest.mark.asyncio + async def test_forwards_okp_filter_to_solr_transport( + self, mocker: MockerFixture + ) -> None: + """The request-level okp filter is forwarded to the Solr transport.""" + mocker.patch( + "utils.vector_search.okp_mcp_available", + new=mocker.AsyncMock(return_value=False), + ) + solr_fetch = mocker.patch( + "utils.vector_search._fetch_okp_rag", + mocker.AsyncMock(return_value=([], [])), + ) + + okp = OkpFilter.model_validate( + {"products": [{"product": "red_hat_enterprise_linux"}]} + ) + client_mock = mocker.AsyncMock() + solr = mocker.Mock() + await _fetch_okp(client_mock, "q", solr=solr, okp=okp) + + solr_fetch.assert_awaited_once_with(client_mock, "q", solr, okp)