From 01138b3af66bf565807667c615a3d816c1322b43 Mon Sep 17 00:00:00 2001 From: dereck Date: Wed, 23 Sep 2026 19:24:46 -0400 Subject: [PATCH] Issue #1272: feat: record the Query Planner's caller from an optional X-LIF-Client header Every query statistics event now carries `client`. The header is optional: absent means "unknown", never a rejected query. A value that is not a short lowercase name is recorded as "invalid" rather than as itself, keeping free text and person data out of the logs. Learner Data Export sends learner-data-export and the MCP server sends semantic-search-mcp. GraphQL forwards its caller's name, or sends graphql, so MCP traffic keeps its origin through the extra hop. The job record remembers the caller for the completion event, since the orchestrator's results callback carries none. Adds graphql_client to the MCP deploy workflow's paths filter; without it, merging would not rebuild the image that sends the new header. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../lif_semantic_search_mcp_server.yml | 1 + CHANGELOG.md | 4 + bases/lif/query_planner_restapi/README.md | 4 + bases/lif/query_planner_restapi/core.py | 19 +-- components/lif/graphql_client/README.md | 2 +- components/lif/graphql_client/core.py | 11 +- components/lif/openapi_to_graphql/README.md | 1 + .../lif/openapi_to_graphql/type_factory.py | 22 +++- components/lif/query_planner_client/core.py | 5 +- components/lif/query_planner_service/core.py | 30 ++++- .../lif/query_planner_service/statistics.py | 39 +++++- .../lif/query_planner_restapi/test_core.py | 46 ++++++++ .../lif/graphql_client/test_core.py | 14 +-- .../test_lif_client_header.py | 111 ++++++++++++++++++ .../lif/query_planner_client/test_core.py | 7 ++ .../lif/query_planner_service/test_core.py | 94 +++++++++++++++ .../query_planner_service/test_statistics.py | 40 +++++++ 17 files changed, 423 insertions(+), 27 deletions(-) create mode 100644 test/components/lif/openapi_to_graphql/test_lif_client_header.py 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 22bc96c3..49691660 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 b9d1faf6..f7a3874b 100644 --- a/components/lif/openapi_to_graphql/README.md +++ b/components/lif/openapi_to_graphql/README.md @@ -22,6 +22,7 @@ from lif.openapi_to_graphql import schema_tools - **`$ref` resolution:** MDR's `generate_openapi_schema` inlines all `$ref`s, so the `$ref` branch in `type_factory.py` exists but isn't exercised by production schemas. Don't delete it — file-based schemas (`USE_OPENAPI_DATA_MODEL_FROM_FILE=true`) still rely on it. - **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. - **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 3d19aab7..95461c77 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"