diff --git a/.github/workflows/lif_semantic_search_mcp_server.yml b/.github/workflows/lif_semantic_search_mcp_server.yml index 915afc64..2463ec6c 100644 --- a/.github/workflows/lif_semantic_search_mcp_server.yml +++ b/.github/workflows/lif_semantic_search_mcp_server.yml @@ -11,6 +11,7 @@ on: - cloudformation/lif-semantic-search-taskdef-includes.yml - components/lif/logging/** - components/lif/datatypes/** + - components/lif/graphql_client/** - components/lif/openapi_schema_parser/** - components/lif/semantic_search_service/** - components/lif/string_utils/** diff --git a/CHANGELOG.md b/CHANGELOG.md index f9754185..721c289a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 completeness" toggle, because the preview is an intentionally partial document. `format` is not validated, matching the runtime translator, which calls `jsonschema.validate` without a `format_checker` +- Query Planner query statistics record the caller as `client`, from an optional `X-LIF-Client` + request header: `unknown` when absent (never a rejected query), `invalid` when not a short + lowercase name. Learner Data Export sends `learner-data-export`, the MCP server + `semantic-search-mcp`, and GraphQL forwards its caller's name or sends `graphql` ### Changed diff --git a/bases/lif/query_planner_restapi/README.md b/bases/lif/query_planner_restapi/README.md index e5f4468e..44e336f1 100644 --- a/bases/lif/query_planner_restapi/README.md +++ b/bases/lif/query_planner_restapi/README.md @@ -24,6 +24,10 @@ The planner reads YAML at startup that describes available information sources ( A **malformed** value for either timeout — `"120s"`, or `0` — stops the service at startup with a message naming the variable, per the convention decided in #1179. Unset or empty falls back to the code default. +## Caller identity + +`POST /query` and `POST /query_async` read an optional `X-LIF-Client` header naming the caller, and every query statistics event (`LIF_QUERY_STATISTICS` log lines, #341) records it as `client` (#1272). It is optional by design: a missing header is recorded as `unknown`, never a rejected query, so the planner works standalone. A value that is not a short lowercase name (`[a-z0-9][a-z0-9._-]{0,63}`) is recorded as `invalid` rather than as itself, which keeps free text and any person data out of the logs. The in-repo callers send `learner-data-export` (`query_planner_client`) and `graphql`; GraphQL forwards its own caller's name instead when it has one, so MCP traffic arrives as `semantic-search-mcp`. + ## Composes - `datatypes` — `LIFQuery`, `LIFRecord`, `LIFUpdate`, planner-side types - `exceptions` diff --git a/bases/lif/query_planner_restapi/core.py b/bases/lif/query_planner_restapi/core.py index 5ddcb672..e1407dff 100644 --- a/bases/lif/query_planner_restapi/core.py +++ b/bases/lif/query_planner_restapi/core.py @@ -2,10 +2,10 @@ import yaml from datetime import datetime from pathlib import Path -from typing import List +from typing import Annotated, List from asyncio import sleep -from fastapi import FastAPI, HTTPException, Response, status +from fastapi import FastAPI, Header, HTTPException, Response, status from lif.datatypes import ( OrchestratorJobResults, @@ -17,6 +17,7 @@ ) from lif.exceptions.core import LIFException from lif.logging.core import get_logger +from lif.query_planner_service import statistics from lif.query_planner_service.core import LIFQueryPlannerService from lif.query_planner_service.datatypes import LIFQueryPlannerConfig, LIFQueryPlannerInfoSourceConfig @@ -144,14 +145,16 @@ def root() -> dict: # temporary, and will be removed soon. # ------------------------------------------------------------------------- @app.post("/query", status_code=status.HTTP_200_OK, response_model=List[LIFRecord]) -async def do_run_query_sync(query: LIFQuery, response: Response) -> List[LIFRecord]: +async def do_run_query_sync( + query: LIFQuery, response: Response, client: Annotated[str | None, Header(alias=statistics.CLIENT_HEADER)] = None +) -> List[LIFRecord]: logger.info("CALL RECEIVED TO /query (sync) API") try: # Counted from before the first run_query: that call makes the cache read and the # orchestrator submission, so starting the clock after it left those round trips # outside the budget entirely (#571). start_time = datetime.now() - result = await service.run_query(query, first_run=True) + result = await service.run_query(query, first_run=True, client=client) if isinstance(result, LIFQueryStatusResponse): logger.info("Query is still processing, entering polling loop") delay_in_seconds: int = MIN_POLLING_DELAY_SECONDS @@ -173,7 +176,7 @@ async def do_run_query_sync(query: LIFQuery, response: Response) -> List[LIFReco result = await service.get_query_status(result.query_id) if result.status == "COMPLETED": logger.info("Query completed successfully, retrieving results") - result = await service.run_query(query, first_run=False) + result = await service.run_query(query, first_run=False, client=client) if isinstance(result, list): logger.info(f"Query completed successfully, returning {len(result)} record(s)") return result @@ -203,10 +206,12 @@ async def do_run_query_sync(query: LIFQuery, response: Response) -> List[LIFReco # to /query in the future. # ------------------------------------------------------------------------- @app.post("/query_async", response_model=List[LIFRecord] | LIFQueryStatusResponse) -async def do_run_query(query: LIFQuery, response: Response) -> List[LIFRecord] | LIFQueryStatusResponse: +async def do_run_query( + query: LIFQuery, response: Response, client: Annotated[str | None, Header(alias=statistics.CLIENT_HEADER)] = None +) -> List[LIFRecord] | LIFQueryStatusResponse: logger.info("CALL RECEIVED TO /query_async API") try: - result = await service.run_query(query, first_run=True) + result = await service.run_query(query, first_run=True, client=client) if isinstance(result, LIFQueryStatusResponse): response.status_code = status.HTTP_202_ACCEPTED response.headers["Location"] = f"/query/{result.query_id}/status" diff --git a/components/lif/graphql_client/README.md b/components/lif/graphql_client/README.md index 9a534bb4..8225ded0 100644 --- a/components/lif/graphql_client/README.md +++ b/components/lif/graphql_client/README.md @@ -8,7 +8,7 @@ Authenticated HTTP client for calling the LIF GraphQL API. Wraps the boilerplate from lif.graphql_client import graphql_query, graphql_mutation, GraphQLClientException ``` -Both functions send `X-API-Key` from `LIF_GRAPHQL_API_KEY` (when set) as the auth header — see CLAUDE.md § "GraphQL API Key Authentication" for the server-side configuration. +Both functions send `X-API-Key` from `LIF_GRAPHQL_API_KEY` (when set) as the auth header — see CLAUDE.md § "GraphQL API Key Authentication" for the server-side configuration. They also always send `X-LIF-Client: semantic-search-mcp`, which GraphQL forwards so the Query Planner's statistics can tell this traffic apart (#1272). | Function | Purpose | |---|---| diff --git a/components/lif/graphql_client/core.py b/components/lif/graphql_client/core.py index 5aaa9f3b..4e625145 100644 --- a/components/lif/graphql_client/core.py +++ b/components/lif/graphql_client/core.py @@ -12,13 +12,18 @@ GRAPHQL_TIMEOUT_READ = float(os.getenv("SEMANTIC_SEARCH_SERVICE__GRAPHQL_TIMEOUT__READ", "300")) +# Names this caller in the Query Planner's query statistics; GraphQL forwards it (#1272). +LIF_CLIENT_NAME = "semantic-search-mcp" + + def _build_headers(api_key: str = "") -> dict: - """Build request headers, including X-API-Key if provided.""" + """Build request headers: X-LIF-Client always, X-API-Key if provided.""" + headers = {"X-LIF-Client": LIF_CLIENT_NAME} if not api_key: api_key = LIF_GRAPHQL_API_KEY if api_key: - return {"X-API-Key": api_key} - return {} + headers["X-API-Key"] = api_key + return headers async def _post(url: str, json: dict, headers: dict, timeout: httpx.Timeout | None = None) -> httpx.Response: diff --git a/components/lif/openapi_to_graphql/README.md b/components/lif/openapi_to_graphql/README.md index 52164179..0853e6cb 100644 --- a/components/lif/openapi_to_graphql/README.md +++ b/components/lif/openapi_to_graphql/README.md @@ -23,6 +23,7 @@ from lif.openapi_to_graphql import schema_tools - **Strawberry `info` typing:** dynamic resolvers must annotate the `info` parameter as `strawberry.types.Info` (not `object` / `Any`). Strawberry 0.297+ identifies the parameter by type, not by name. - **Backend failures raise:** the root query resolver raises on a non-200 from the Query Planner so Strawberry emits an `errors` entry, matching the update mutation. Returning `[]` made every failure indistinguishable from a learner with no data (#1264). - **Field name preservation:** uses `strawberry.field(name=field_name)` so the wire shape preserves PascalCase entity / camelCase scalar conventions ([`docs/specs/data-model-rules.md`](../../../docs/specs/data-model-rules.md)). +- **Caller forwarding:** the root query resolver forwards the incoming `X-LIF-Client` header to the Query Planner, or sends `graphql` when there is none, so the planner's statistics see the original caller (#1272). It reads the request from `info.context["request"]`, which Strawberry's FastAPI router provides; it does not validate the value, since the planner does. ## Used by - `bases/lif/api_graphql` — single consumer; the GraphQL service's whole reason for existing is this component. diff --git a/components/lif/openapi_to_graphql/type_factory.py b/components/lif/openapi_to_graphql/type_factory.py index 1018f04f..af4ff83c 100644 --- a/components/lif/openapi_to_graphql/type_factory.py +++ b/components/lif/openapi_to_graphql/type_factory.py @@ -54,6 +54,13 @@ os.getenv("LIF_GRAPHQL_CLIENT_TIMEOUT_SECONDS") or os.getenv("LIF_QUERY_TIMEOUT_SECONDS") or "20" ) +# Callers name themselves to the Query Planner's statistics in this header (#1272). GraphQL +# forwards the name it received, so a query that arrives through here -- the MCP server's -- +# keeps its origin; it names itself only when its own caller did not. Validating the value is +# the planner's job, not this relay's. +LIF_CLIENT_HEADER = "X-LIF-Client" +LIF_CLIENT_NAME = "graphql" + # === Constants === @@ -768,6 +775,19 @@ def create_input_type( return create_nested_input_type(type_name, schema, openapi, created_types, input_type_cache) +def lif_client_headers(info: Any) -> Dict[str, str]: + """ + The X-LIF-Client header to send the Query Planner: the incoming one, else this service's name. + + Strawberry's FastAPI router puts the request in `info.context["request"]`; a schema + executed without one (tests, scripts) simply names itself. + """ + context = info.context if isinstance(info.context, dict) else {} + request = context.get("request") + incoming = request.headers.get(LIF_CLIENT_HEADER) if request is not None else None + return {LIF_CLIENT_HEADER: incoming or LIF_CLIENT_NAME} + + # === Root Query Type Construction === @@ -822,7 +842,7 @@ async def resolver(self: Any, info: Any, filter: Optional[Any] = None) -> List[A logger.info(f"Query: {query}") # Make the backend API call async with httpx.AsyncClient(timeout=httpx.Timeout(LIF_GRAPHQL_CLIENT_TIMEOUT_SECONDS)) as client: - response = await client.post(query_planner_query_url, json=query) + response = await client.post(query_planner_query_url, json=query, headers=lif_client_headers(info)) if response.status_code == 200: response_json = response.json() diff --git a/components/lif/query_planner_client/core.py b/components/lif/query_planner_client/core.py index 51c3656c..4c7156c9 100644 --- a/components/lif/query_planner_client/core.py +++ b/components/lif/query_planner_client/core.py @@ -10,6 +10,9 @@ # Default timeout for Query Planner client API calls (in seconds) DEFAULT_QUERY_PLANNER_CLIENT_TIMEOUT_SECONDS = 30 +# Names this caller in the Query Planner's query statistics (#1272). +LIF_CLIENT_HEADERS = {"X-LIF-Client": "learner-data-export"} + def _get_query_planner_timeout_seconds() -> int: return int(os.getenv("QUERY_PLANNER_CLIENT_TIMEOUT_SECONDS", str(DEFAULT_QUERY_PLANNER_CLIENT_TIMEOUT_SECONDS))) @@ -39,7 +42,7 @@ async def fetch_query_from_query_planner(base_url: str, query: dict) -> list[dic try: async for client in _get_query_planner_client(): - response = await client.post(url, json=query) + response = await client.post(url, json=query, headers=LIF_CLIENT_HEADERS) except httpx.TimeoutException as e: msg = f"Query Planner request timed out due to: {e}" logger.error(msg) diff --git a/components/lif/query_planner_service/core.py b/components/lif/query_planner_service/core.py index c8486a48..a9ccf131 100644 --- a/components/lif/query_planner_service/core.py +++ b/components/lif/query_planner_service/core.py @@ -64,6 +64,7 @@ def _emit_query_planned( paths_not_in_cache: List[str], lif_query_plan: LIFQueryPlan | None = None, correlation_id: str | None = None, + client: str = statistics.CLIENT_UNKNOWN, ) -> None: """ Emit the planning-phase statistics event. Never raises -- statistics must not fail a query. @@ -72,7 +73,7 @@ def _emit_query_planned( logger.info( statistics.format_event( statistics.build_query_planned_event( - outcome, requested_paths, paths_not_in_cache, lif_query_plan, correlation_id + outcome, requested_paths, paths_not_in_cache, lif_query_plan, correlation_id, client ) ) ) @@ -81,12 +82,15 @@ def _emit_query_planned( # Main function to run a query # ------------------------------------------------------------------------- - async def run_query(self, query: LIFQuery, first_run: bool) -> List[LIFRecord] | LIFQueryStatusResponse: + async def run_query( + self, query: LIFQuery, first_run: bool, client: str | None = None + ) -> List[LIFRecord] | LIFQueryStatusResponse: """ Execute a LIF query. Args: query (LIFQuery): Input query with filter and selected fields. + client (str | None): The raw `X-LIF-Client` header value, or None when absent. Returns: List[LIFRecord]: List of matching LIF records (persons) from the database, with only @@ -95,6 +99,8 @@ async def run_query(self, query: LIFQuery, first_run: bool) -> List[LIFRecord] | Raises: LIFException: If the query fails. """ + # Reduced here, at the service boundary, so no raw header value reaches a log line. + client = statistics.normalize_client(client) try: # Send the query to the LIF Cache service lif_records: List[LIFRecord] = await query_lif_cache( @@ -120,7 +126,9 @@ async def run_query(self, query: LIFQuery, first_run: bool) -> List[LIFRecord] | # Guarded on first_run: the sync /query path calls run_query twice, and the # second call lands here once the orchestrator has filled the cache (#341 gap 3). if first_run: - self._emit_query_planned(statistics.OUTCOME_SERVED_FROM_CACHE, lif_fragment_paths, []) + self._emit_query_planned( + statistics.OUTCOME_SERVED_FROM_CACHE, lif_fragment_paths, [], client=client + ) return lif_records logger.info(f"LIF Record does not contain all requested fields, missing: {lif_fragment_paths_not_found}") @@ -140,7 +148,10 @@ async def run_query(self, query: LIFQuery, first_run: bool) -> List[LIFRecord] | f"No information sources found for the requested LIF fragment paths: {lif_fragment_paths}. Returning LIF records found in cache." ) self._emit_query_planned( - statistics.OUTCOME_NO_SOURCES_AVAILABLE, lif_fragment_paths, lif_fragment_paths_not_found + statistics.OUTCOME_NO_SOURCES_AVAILABLE, + lif_fragment_paths, + lif_fragment_paths_not_found, + client=client, ) return lif_records @@ -162,11 +173,12 @@ async def run_query(self, query: LIFQuery, first_run: bool) -> List[LIFRecord] | lif_fragment_paths, lif_fragment_paths_not_found, lif_query_plan, + client=client, ) return lif_records lif_query_planner_job = LIFQueryPlannerJob( - job_id=orchestrator_job_request_response.run_id, query=query, status="PENDING" + job_id=orchestrator_job_request_response.run_id, query=query, status="PENDING", client=client ) prune_job_store() @@ -179,6 +191,7 @@ async def run_query(self, query: LIFQuery, first_run: bool) -> List[LIFRecord] | lif_fragment_paths_not_found, lif_query_plan, lif_query_planner_job.job_id, + client, ) query_status_response = LIFQueryStatusResponse(query_id=lif_query_planner_job.job_id, status="PENDING") return query_status_response @@ -304,7 +317,9 @@ async def run_post_orchestration_results(self, results: OrchestratorJobResults) try: logger.info( - statistics.format_event(statistics.build_query_completed_event(results, lif_fragment_paths)) + statistics.format_event( + statistics.build_query_completed_event(results, lif_fragment_paths, job.client) + ) ) except Exception: logger.exception("Failed to emit query statistics") @@ -408,6 +423,7 @@ class LIFQueryPlannerJob(BaseModel): status (str): Status of the job (e.g., 'pending', 'running', 'completed'). created_timestamp (str): Timestamp of when the job was created. updated_timestamp (str): Timestamp of when the job was last updated. + client (str): The caller that submitted the query, for the completion statistics event. """ job_id: str = Field(..., description="Unique identifier for the job") @@ -423,6 +439,8 @@ class LIFQueryPlannerJob(BaseModel): description="Timestamp of when the job was last updated", default_factory=lambda: datetime.now(timezone.utc).isoformat(), ) + # The orchestrator's results callback carries no caller, so the job remembers it (#1272). + client: str = Field(statistics.CLIENT_UNKNOWN, description="The caller that submitted the query") # ------------------------------------------------------------------------- diff --git a/components/lif/query_planner_service/statistics.py b/components/lif/query_planner_service/statistics.py index 4bfd431d..582a209f 100644 --- a/components/lif/query_planner_service/statistics.py +++ b/components/lif/query_planner_service/statistics.py @@ -10,6 +10,7 @@ """ import json +import re from typing import Dict, List from lif.datatypes.core import LIFQueryPlan @@ -26,6 +27,35 @@ OUTCOME_ORCHESTRATOR_SUBMISSION_FAILED: str = "orchestrator_submission_failed" OUTCOME_ORCHESTRATED: str = "orchestrated" +# Caller identity (#1272). Callers name themselves in this optional request header; a missing +# header is "unknown", never a rejected query, so the planner keeps working standalone (ADR 0004). +CLIENT_HEADER: str = "X-LIF-Client" +CLIENT_UNKNOWN: str = "unknown" +CLIENT_INVALID: str = "invalid" +# A short lowercase name, nothing else. The value is caller-supplied and lands in a statistics +# dimension, so anything outside this shape is recorded as "invalid" rather than as itself: +# that keeps free text -- and any person data a caller might put there (#1269) -- out of the +# logs, and keeps the dimension's cardinality bounded. +_CLIENT_PATTERN = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}") + + +def normalize_client(raw: str | None) -> str: + """ + Reduce a caller-supplied `X-LIF-Client` value to a safe statistics dimension. + + Args: + raw (str | None): The header value as received, or None when absent. + + Returns: + str: The value itself when it is a well-formed client name, CLIENT_UNKNOWN when it is + absent or empty, and CLIENT_INVALID otherwise. + """ + if not raw: + return CLIENT_UNKNOWN + if _CLIENT_PATTERN.fullmatch(raw): + return raw + return CLIENT_INVALID + def _normalize_path(path: str) -> str: """ @@ -52,6 +82,7 @@ def build_query_planned_event( paths_not_in_cache: List[str], lif_query_plan: LIFQueryPlan | None = None, correlation_id: str | None = None, + client: str = CLIENT_UNKNOWN, ) -> Dict: """ Build the statistics event for the planning phase of a query. @@ -62,6 +93,7 @@ def build_query_planned_event( paths_not_in_cache (List[str]): Of those, the ones the cache could not answer. lif_query_plan (LIFQueryPlan | None): The plan, when one was built. correlation_id (str | None): The orchestrator run id, when one was obtained. + client (str): The caller, already reduced by `normalize_client`. Returns: Dict: The event. Contains no person data. @@ -69,6 +101,7 @@ def build_query_planned_event( parts = lif_query_plan.root if lif_query_plan else [] return { "event": "query_planned", + "client": client, "outcome": outcome, "correlation_id": correlation_id, "requested_paths": sorted(_normalize_path(path) for path in requested_paths), @@ -86,13 +119,16 @@ def build_query_planned_event( } -def build_query_completed_event(results: OrchestratorJobResults, requested_paths: List[str]) -> Dict: +def build_query_completed_event( + results: OrchestratorJobResults, requested_paths: List[str], client: str = CLIENT_UNKNOWN +) -> Dict: """ Build the statistics event for the orchestration results of a query. Args: results (OrchestratorJobResults): The results posted back by the Orchestrator. requested_paths (List[str]): LIF fragment paths the original query asked for. + client (str): The caller of the original query, already reduced by `normalize_client`. Returns: Dict: The event. Contains no person data -- fragment paths, not fragments. @@ -106,6 +142,7 @@ def build_query_completed_event(results: OrchestratorJobResults, requested_paths normalized_requested = {_normalize_path(path) for path in requested_paths} return { "event": "query_completed", + "client": client, "correlation_id": results.run_id, "requested_paths": sorted(normalized_requested), "fulfilled_paths": sorted(fulfilled_paths), diff --git a/test/bases/lif/query_planner_restapi/test_core.py b/test/bases/lif/query_planner_restapi/test_core.py index e156ac26..8637fa20 100644 --- a/test/bases/lif/query_planner_restapi/test_core.py +++ b/test/bases/lif/query_planner_restapi/test_core.py @@ -421,3 +421,49 @@ def test_async_query_endpoint_does_not_log_returned_records(caplog): assert "Bellwether" not in caplog.text assert "Sentinel" not in caplog.text assert "Query completed successfully" in caplog.text + + +# ------------------------------------------------------------------------- +# #1272 — the optional X-LIF-Client header, through the real HTTP layer. +# ------------------------------------------------------------------------- +_FULL_CACHE_RECORD = {"person": [{"name": [{"givenName": ["John"], "familyName": "Doe"}]}]} + + +def _statistics_events(caplog) -> list: + from lif.query_planner_service import statistics + + prefix = statistics.QUERY_STATISTICS_PREFIX + " " + return [json.loads(line[line.index(prefix) + len(prefix) :]) for line in caplog.text.splitlines() if prefix in line] + + +def _post_query(path: str, headers: dict, caplog) -> tuple[int, list]: + """POST a query through the app with the real service; only the planner's own outbound calls are faked.""" + from fastapi.testclient import TestClient + + from lif.query_planner_restapi import core + + cache_response = MagicMock(status_code=200) + cache_response.json.return_value = [_FULL_CACHE_RECORD] + cache_response.raise_for_status.return_value = None + body = _make_query().model_dump(mode="json", by_alias=True) + with patch("httpx.AsyncClient.post", AsyncMock(return_value=cache_response)), caplog.at_level(logging.INFO): + response = TestClient(core.app).post(path, json=body, headers=headers) + return response.status_code, _statistics_events(caplog) + + +@patch.dict(os.environ, _ENV) +def test_query_without_the_client_header_succeeds_and_still_emits_statistics(caplog): + for path in ["/query", "/query_async"]: + caplog.clear() + status_code, events = _post_query(path, {}, caplog) + assert status_code == 200, path + assert [e["client"] for e in events] == ["unknown"], path + + +@patch.dict(os.environ, _ENV) +def test_query_records_the_client_header(caplog): + for path in ["/query", "/query_async"]: + caplog.clear() + status_code, events = _post_query(path, {"X-LIF-Client": "learner-data-export"}, caplog) + assert status_code == 200, path + assert [e["client"] for e in events] == ["learner-data-export"], path diff --git a/test/components/lif/graphql_client/test_core.py b/test/components/lif/graphql_client/test_core.py index 8f446d8e..088f1dbb 100644 --- a/test/components/lif/graphql_client/test_core.py +++ b/test/components/lif/graphql_client/test_core.py @@ -32,7 +32,7 @@ async def test_graphql_query_with_api_key(mock_post): assert result == {"data": {"person": []}} mock_post.assert_called_once() call_kwargs = mock_post.call_args - assert call_kwargs.kwargs["headers"] == {"X-API-Key": "test-key-123"} + assert call_kwargs.kwargs["headers"] == {"X-LIF-Client": "semantic-search-mcp", "X-API-Key": "test-key-123"} @patch("lif.graphql_client.core._post") @@ -47,7 +47,7 @@ async def test_graphql_query_without_api_key(mock_post): assert result == {"data": {"person": []}} call_kwargs = mock_post.call_args - assert call_kwargs.kwargs["headers"] == {} + assert call_kwargs.kwargs["headers"] == {"X-LIF-Client": "semantic-search-mcp"} @patch("lif.graphql_client.core._post") @@ -62,7 +62,7 @@ async def test_graphql_query_env_var_fallback(mock_post): assert result == {"data": {"person": []}} call_kwargs = mock_post.call_args - assert call_kwargs.kwargs["headers"] == {"X-API-Key": "env-key-456"} + assert call_kwargs.kwargs["headers"] == {"X-LIF-Client": "semantic-search-mcp", "X-API-Key": "env-key-456"} @patch("lif.graphql_client.core._post") @@ -93,7 +93,7 @@ async def test_graphql_mutation_with_api_key(mock_post): assert result == {"data": {"updatePerson": {}}} call_kwargs = mock_post.call_args - assert call_kwargs.kwargs["headers"] == {"X-API-Key": "test-key-123"} + assert call_kwargs.kwargs["headers"] == {"X-LIF-Client": "semantic-search-mcp", "X-API-Key": "test-key-123"} @patch("lif.graphql_client.core._post") @@ -133,19 +133,19 @@ async def test_graphql_query_custom_timeout(mock_post): def test_build_headers_with_key(): headers = core._build_headers(api_key="my-key") - assert headers == {"X-API-Key": "my-key"} + assert headers == {"X-LIF-Client": "semantic-search-mcp", "X-API-Key": "my-key"} def test_build_headers_without_key(): with mock.patch.object(core, "LIF_GRAPHQL_API_KEY", ""): headers = core._build_headers(api_key="") - assert headers == {} + assert headers == {"X-LIF-Client": "semantic-search-mcp"} def test_build_headers_env_var_fallback(): with mock.patch.object(core, "LIF_GRAPHQL_API_KEY", "env-key"): headers = core._build_headers(api_key="") - assert headers == {"X-API-Key": "env-key"} + assert headers == {"X-LIF-Client": "semantic-search-mcp", "X-API-Key": "env-key"} def test_graphql_client_exception(): diff --git a/test/components/lif/openapi_to_graphql/test_lif_client_header.py b/test/components/lif/openapi_to_graphql/test_lif_client_header.py new file mode 100644 index 00000000..79a87728 --- /dev/null +++ b/test/components/lif/openapi_to_graphql/test_lif_client_header.py @@ -0,0 +1,111 @@ +""" +GraphQL relays the caller's X-LIF-Client to the Query Planner (#1272). + +The MCP server reaches the planner only through GraphQL, so if GraphQL named itself on every +query the planner could never tell MCP traffic from any other GraphQL traffic. These tests +drive a real FastAPI app with Strawberry's GraphQLRouter -- mounted the way +`bases/lif/api_graphql/core.py` mounts it -- because the incoming request only reaches the +resolver through the router's default context. A schema executed directly never sees it. +""" + +from fastapi import FastAPI +from fastapi.testclient import TestClient +from strawberry.fastapi import GraphQLRouter + +from lif.openapi_to_graphql import type_factory +from lif.openapi_to_graphql.core import generate_graphql_schema + +PERSON_QUERY = '{ person(filter: {name: "x"}) { name } }' + + +class _CapturingResponse: + status_code = 200 + text = "" + + def json(self): + return [{"person": []}] + + +class _CapturingAsyncClient: + """Stands in for httpx.AsyncClient and records the headers of every POST to the planner.""" + + def __init__(self): + self.sent_headers: list[dict] = [] + + def __call__(self, *args, **kwargs): + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *args): + return False + + async def post(self, url, json=None, headers=None): + self.sent_headers.append(headers or {}) + return _CapturingResponse() + + +async def _schema(monkeypatch): + # type_factory.input_type_cache is module-level and keyed by type name only, so a + # PersonInput built by another test in the same process would be reused here. + monkeypatch.setattr(type_factory, "input_type_cache", {}) + openapi = { + "components": { + "schemas": { + "Person": { + "type": "array", + "properties": { + "name": {"DataType": "xsd:string", "Array": "No", "x-queryable": True, "x-mutable": False} + }, + } + } + } + } + return await generate_graphql_schema( + openapi=openapi, + root_type_name="Person", + query_planner_query_url="http://localhost:9999/query", + query_planner_update_url="http://localhost:9999/update", + ) + + +async def _app(monkeypatch) -> FastAPI: + app = FastAPI() + app.include_router(GraphQLRouter(await _schema(monkeypatch), prefix="/graphql")) + return app + + +async def _headers_sent_to_the_planner(monkeypatch, incoming: dict) -> dict: + app = await _app(monkeypatch) + fake = _CapturingAsyncClient() + monkeypatch.setattr(type_factory.httpx, "AsyncClient", fake) + response = TestClient(app).post("/graphql", json={"query": PERSON_QUERY}, headers=incoming) + assert response.status_code == 200 and "errors" not in response.json(), response.text + assert len(fake.sent_headers) == 1 + return fake.sent_headers[0] + + +async def test_forwards_the_callers_client_name(monkeypatch): + sent = await _headers_sent_to_the_planner(monkeypatch, {"X-LIF-Client": "semantic-search-mcp"}) + assert sent["X-LIF-Client"] == "semantic-search-mcp" + + +async def test_names_itself_when_the_caller_did_not(monkeypatch): + sent = await _headers_sent_to_the_planner(monkeypatch, {}) + assert sent["X-LIF-Client"] == "graphql" + + +async def test_relays_a_malformed_name_unchanged_for_the_planner_to_reject(monkeypatch): + """Validation is the planner's, so it happens once, where the value is recorded.""" + sent = await _headers_sent_to_the_planner(monkeypatch, {"X-LIF-Client": "Not A Name"}) + assert sent["X-LIF-Client"] == "Not A Name" + + +async def test_a_schema_executed_without_a_request_still_names_itself(monkeypatch): + schema = await _schema(monkeypatch) + fake = _CapturingAsyncClient() + monkeypatch.setattr(type_factory.httpx, "AsyncClient", fake) + result = await schema.execute(PERSON_QUERY) + assert not result.errors, result.errors + assert fake.sent_headers == [{"X-LIF-Client": "graphql"}] diff --git a/test/components/lif/query_planner_client/test_core.py b/test/components/lif/query_planner_client/test_core.py index 44496811..9e81fd44 100644 --- a/test/components/lif/query_planner_client/test_core.py +++ b/test/components/lif/query_planner_client/test_core.py @@ -59,6 +59,13 @@ async def test_query_sent_as_json_body(): assert mock_client.post.call_args[1]["json"] == _QUERY +async def test_query_names_this_caller_to_the_query_planner(): + mock_cls, mock_client = _make_http_mock(200, []) + with mock.patch("lif.query_planner_client.core.httpx.AsyncClient", mock_cls): + await fetch_query_from_query_planner(_BASE_URL, _QUERY) + assert mock_client.post.call_args[1]["headers"]["X-LIF-Client"] == "learner-data-export" + + async def test_non_200_raises_query_planner_exception(): mock_cls, _ = _make_http_mock(500) with mock.patch("lif.query_planner_client.core.httpx.AsyncClient", mock_cls): diff --git a/test/components/lif/query_planner_service/test_core.py b/test/components/lif/query_planner_service/test_core.py index be4cdc17..0c6a4b07 100644 --- a/test/components/lif/query_planner_service/test_core.py +++ b/test/components/lif/query_planner_service/test_core.py @@ -607,6 +607,100 @@ async def run_both(): assert events[0]["outcome"] == statistics.OUTCOME_ORCHESTRATED +@patch("httpx.AsyncClient.post") +def test_every_planned_outcome_records_the_client(mock_post, caplog): + """ + Each of the four outcomes emits from its own call site, so each must be handed the + client -- one forgotten site would silently bucket that outcome under "unknown" (#1272). + """ + cache_miss = _create_mock_post_response(200, [{"person": [{}]}], "https://api.example.com/query") + full_cache = _create_mock_post_response(200, [_FULL_CACHE_RECORD], "https://api.example.com/query") + submitted = _create_mock_post_response(200, {"run_id": "run-1"}, "https://api.example.com/jobs") + no_sources = core.LIFQueryPlannerService( + config=core.LIFQueryPlannerConfig( + lif_cache_url="https://api.example.com", + lif_orchestrator_url="https://api.example.com", + information_sources_config=[], + ) + ) + + async def run_all(): + mock_post.side_effect = [full_cache] + await _stats_service().run_query(_sentinel_query(), first_run=True, client="graphql") + mock_post.side_effect = [cache_miss] + await no_sources.run_query(_sentinel_query(), first_run=True, client="graphql") + mock_post.side_effect = [cache_miss, httpx.ConnectError("down")] + await _stats_service().run_query(_sentinel_query(), first_run=True, client="graphql") + mock_post.side_effect = [cache_miss, submitted] + await _stats_service().run_query(_sentinel_query(), first_run=True, client="graphql") + + with patch.object(core, "JOB_STORE", {}), caplog.at_level(logging.INFO): + asyncio.run(run_all()) + + events = _emitted_events(caplog) + assert [e["outcome"] for e in events] == [ + statistics.OUTCOME_SERVED_FROM_CACHE, + statistics.OUTCOME_NO_SOURCES_AVAILABLE, + statistics.OUTCOME_ORCHESTRATOR_SUBMISSION_FAILED, + statistics.OUTCOME_ORCHESTRATED, + ] + assert {e["client"] for e in events} == {"graphql"} + + +@patch("httpx.AsyncClient.post") +def test_run_query_without_a_client_still_succeeds_and_emits_unknown(mock_post, caplog): + """The header is optional: its absence is a statistic, never a failed query (ADR 0004).""" + mock_post.side_effect = [_create_mock_post_response(200, [_FULL_CACHE_RECORD], "https://api.example.com/query")] + + with patch.object(core, "JOB_STORE", {}), caplog.at_level(logging.INFO): + records = asyncio.run(_stats_service().run_query(_sentinel_query(), first_run=True)) + + assert len(records) == 1 + events = _emitted_events(caplog) + assert len(events) == 1 + assert events[0]["client"] == statistics.CLIENT_UNKNOWN + + +@patch("httpx.AsyncClient.post") +def test_a_malformed_client_is_recorded_as_invalid_and_never_logged_raw(mock_post, caplog): + mock_post.side_effect = [_create_mock_post_response(200, [_FULL_CACHE_RECORD], "https://api.example.com/query")] + raw = "Sentinel-1234 john.doe@example.edu" + + with patch.object(core, "JOB_STORE", {}), caplog.at_level(logging.DEBUG): + asyncio.run(_stats_service().run_query(_sentinel_query(), first_run=True, client=raw)) + + assert _emitted_events(caplog)[0]["client"] == statistics.CLIENT_INVALID + assert "john.doe" not in caplog.text + + +@patch("httpx.AsyncClient.post") +def test_the_client_survives_to_the_completed_event(mock_post, caplog): + """ + The orchestrator's results callback carries no caller, so the only way the completed + event can name one is the job remembering it from the original query. + """ + mock_post.side_effect = [ + _create_mock_post_response(200, [{"person": [{}]}], "https://api.example.com/query"), + _create_mock_post_response(200, {"run_id": "run-1"}, "https://api.example.com/jobs"), + _create_mock_post_response(200, {}, "https://api.example.com/save"), + ] + results = OrchestratorJobResults(run_id="run-1", query_plan_part_results=[]) + service = _stats_service() + + async def submit_then_complete(): + await service.run_query(_sentinel_query(), first_run=True, client="learner-data-export") + await service.run_post_orchestration_results(results) + + with patch.object(core, "JOB_STORE", {}), caplog.at_level(logging.INFO): + asyncio.run(submit_then_complete()) + + events = _emitted_events(caplog) + assert [(e["event"], e["client"]) for e in events] == [ + ("query_planned", "learner-data-export"), + ("query_completed", "learner-data-export"), + ] + + @patch("httpx.AsyncClient.post") def test_run_post_orchestration_results_emits_completed_statistics(mock_post, caplog): mock_post.return_value = _create_mock_post_response(200, {}, "https://api.example.com/save") diff --git a/test/components/lif/query_planner_service/test_statistics.py b/test/components/lif/query_planner_service/test_statistics.py index a836c0d9..ae31d868 100644 --- a/test/components/lif/query_planner_service/test_statistics.py +++ b/test/components/lif/query_planner_service/test_statistics.py @@ -117,3 +117,43 @@ def test_build_query_completed_event_matches_paths_across_the_casing_boundary(): assert event["fulfilled_paths"] == ["Person.name"] assert event["paths_not_fulfilled"] == [] + + +# ------------------------------------------------------------------------- +# #1272 — the caller, from the optional X-LIF-Client header. +# ------------------------------------------------------------------------- +def test_normalize_client_treats_a_missing_or_empty_header_as_unknown(): + assert statistics.normalize_client(None) == statistics.CLIENT_UNKNOWN + assert statistics.normalize_client("") == statistics.CLIENT_UNKNOWN + + +def test_normalize_client_keeps_a_well_formed_name(): + for name in ["graphql", "learner-data-export", "semantic-search-mcp", "org1.lde_v2", "a" * 64]: + assert statistics.normalize_client(name) == name + + +def test_normalize_client_records_anything_else_as_invalid_rather_than_as_itself(): + # Each of these would otherwise land verbatim in a statistics dimension: free text, a + # person identifier smuggled into the header (#1269), a log-line injection, unbounded length. + for raw in [ + "GraphQL", + "-leading-dash", + "a" * 65, + "Sentinel-1234 john.doe@example.edu", + 'graphql"} {"injected": true', + "graphql\nLIF_QUERY_STATISTICS {}", + " ", + ]: + assert statistics.normalize_client(raw) == statistics.CLIENT_INVALID, raw + + +def test_both_events_carry_the_client_and_default_to_unknown(): + planned = statistics.build_query_planned_event(statistics.OUTCOME_SERVED_FROM_CACHE, ["person.name"], []) + completed = statistics.build_query_completed_event(_results(), requested_paths=["person.name"]) + assert planned["client"] == completed["client"] == statistics.CLIENT_UNKNOWN + + planned = statistics.build_query_planned_event( + statistics.OUTCOME_SERVED_FROM_CACHE, ["person.name"], [], client="graphql" + ) + completed = statistics.build_query_completed_event(_results(), requested_paths=["person.name"], client="graphql") + assert planned["client"] == completed["client"] == "graphql"