fix(mcs): send credentials to the measure engine and target the job's MCS

backend/app/services/fhir_client.py

@@ -767,6 +767,7 @@ async def evaluate_measure(
     period_start: str,
     period_end: str,
     measure_engine_url: str | None = None,
+    auth_headers: dict[str, str] | None = None,
 ) -> dict[str, Any]:
     """Call $evaluate-measure on the measure engine for a single patient.
 
@@ -776,6 +777,9 @@ async def evaluate_measure(
             haven't yet been wired to per-job active-MCS context. The
             orchestrator passes `job.mcs_url` so jobs run against the MCS
             that was active at job creation time, not whatever's active now.
+        auth_headers: Credentials for the MCS, resolved from the job's linked
+            MCSConfig. Omitted for unauthenticated engines (local HAPI). A
+            remote MCS rejects every evaluation with 401 without these.
 
     Raises FhirOperationError (with the MCS OperationOutcome preserved) on
     4xx/5xx responses and on 200 OK where the body is an OperationOutcome
@@ -796,7 +800,7 @@ async def evaluate_measure(
                 extra={"measure_id": measure_id, "patient_id": patient_id, "attempt": attempt + 1},
             )
             start_ms = int(time.monotonic() * 1000)
-            resp = await client.get(url)
+            resp = await client.get(url, headers=auth_headers or {})
             latency_ms = int(time.monotonic() * 1000) - start_ms
             try:
                 resp.raise_for_status()
@@ -918,7 +922,7 @@ async def delete_measure(measure_id: str) -> None:
             resp.raise_for_status()
 
 
-async def wipe_patient_data(*, base_url: str, strict: bool = True) -> None:
+async def wipe_patient_data(*, base_url: str, strict: bool = True, auth_headers: dict[str, str] | None = None) -> None:
     """Delete patient-related data from a FHIR server.
 
     Called at the START of a new job (with base_url=MEASURE_ENGINE_URL) to clean
@@ -929,6 +933,10 @@ async def wipe_patient_data(*, base_url: str, strict: bool = True) -> None:
     A timed-out DELETE leaves HAPI's server-side operation still running; pushing new
     data over it causes the in-flight DELETE to wipe the freshly-pushed resources.
     The strict parameter is kept for API compatibility but no longer silences failures.
+
+    `auth_headers` credentials the target server. A remote MCS rejects every DELETE
+    with 401 without them, which would leave the prior job's patients in place and
+    contaminate the next evaluation.
     """
     # Delete clinical resources before Patient: HAPI returns 409 when Patient is
     # referenced by Condition/Encounter/etc., so Patient must be last.
@@ -965,7 +973,7 @@ async def wipe_patient_data(*, base_url: str, strict: bool = True) -> None:
             try:
                 # Use conditional delete: DELETE ResourceType?_lastUpdated=gt1900-01-01
                 delete_url = f"{base_url}/{rt}?_lastUpdated=gt1900-01-01"
-                resp = await client.delete(delete_url)
+                resp = await client.delete(delete_url, headers=auth_headers or {})
                 if resp.status_code < 300:
                     logger.info("Wiped resource type", extra={"resourceType": rt})
                 else:
@@ -1036,21 +1044,36 @@ async def _delete_all_of_type(client: httpx.AsyncClient, resource_type: str, bas
                 break
 
 
-async def resolve_evaluated_resource(reference: str) -> dict[str, Any]:
-    """Resolve an evaluatedResource reference from the measure engine."""
-    url = f"{settings.MEASURE_ENGINE_URL}/{reference}"
+async def resolve_evaluated_resource(
+    reference: str,
+    base_url: str | None = None,
+    auth_headers: dict[str, str] | None = None,
+) -> dict[str, Any]:
+    """Resolve an evaluatedResource reference from the measure engine.
+
+    `base_url`/`auth_headers` target the job's MCS. They default to the env-var
+    engine with no credentials, which only works for a local unauthenticated HAPI.
+    """
+    url = f"{base_url or settings.MEASURE_ENGINE_URL}/{reference}"
     async with httpx.AsyncClient(timeout=30.0) as client:
-        resp = await client.get(url)
+        resp = await client.get(url, headers=auth_headers or {})
         resp.raise_for_status()
         return resp.json()
 
 
-async def snapshot_evaluated_resources(measure_report: dict[str, Any]) -> list[dict[str, Any]] | None:
+async def snapshot_evaluated_resources(
+    measure_report: dict[str, Any],
+    base_url: str | None = None,
+    auth_headers: dict[str, str] | None = None,
+) -> list[dict[str, Any]] | None:
     """Resolve every evaluatedResource reference in a MeasureReport to a stored snapshot.
 
     Returns a list of full FHIR resources. Per-reference failures are skipped and logged
     rather than raised — partial snapshots are still useful and the caller has already
     persisted the MeasureReport itself. Returns None when there is nothing to snapshot.
+
+    `base_url`/`auth_headers` identify the MCS the report came from, so snapshots
+    are read back from the same server that evaluated them.
     """
     refs = [
         r.get("reference")
@@ -1063,7 +1086,7 @@ async def snapshot_evaluated_resources(measure_report: dict[str, Any]) -> list[d
     resources: list[dict[str, Any]] = []
     for ref in refs:
         try:
-            resources.append(await resolve_evaluated_resource(ref))
+            resources.append(await resolve_evaluated_resource(ref, base_url, auth_headers))
         except Exception as exc:
             logger.warning(
                 "Failed to snapshot evaluated resource",

backend/app/services/orchestrator.py

@@ -18,6 +18,7 @@
 from app.db import async_session
 from app.models.config import CDRConfig
 from app.models.job import Batch, BatchStatus, Job, JobStatus, MeasureResult
+from app.models.mcs_config import MCSConfig
 from app.services.fhir_client import (
     BatchQueryStrategy,
     DataRequirementsStrategy,
@@ -171,9 +172,16 @@ async def run_job(job_id: int) -> None:
         await session.commit()
 
     try:
+        # Resolve the MCS up front: the wipe below must target the same engine the
+        # job will push to and evaluate against, not the env-var default. Pointing
+        # the wipe at a different server would leave the real target's prior-run
+        # data in place and silently contaminate this job's populations.
+        mcs_url = await _get_mcs_url(job_id)
+        mcs_auth_headers = await _get_mcs_auth_headers(job_id)
+
         # Step 1: Wipe patient data from measure engine (cleanup from prior job)
         logger.info("Wiping prior patient data from measure engine", extra={"job_id": job_id})
-        await wipe_patient_data(base_url=settings.MEASURE_ENGINE_URL, strict=False)
+        await wipe_patient_data(base_url=mcs_url, strict=False, auth_headers=mcs_auth_headers)
         if await _stop_or_delete_job(job_id):
             return
 
@@ -238,9 +246,8 @@ async def run_job(job_id: int) -> None:
         # Step 5: Process batches with concurrency control
         semaphore = asyncio.Semaphore(settings.MAX_WORKERS)
 
-        # Resolve MCS URL once for the whole job. Falls back to the env-var
-        # default if Job.mcs_url is NULL (legacy rows pre-dating PR #4).
-        mcs_url = await _get_mcs_url(job_id)
+        # mcs_url / mcs_auth_headers were resolved before the wipe above so every
+        # MCS interaction in this job targets one server with one set of credentials.
 
         async def process_batch(batch_id: int) -> None:
             async with semaphore:
@@ -251,6 +258,7 @@ async def process_batch(batch_id: int) -> None:
                     cdr_url=cdr_url,
                     auth_headers=auth_headers,
                     mcs_url=mcs_url,
+                    mcs_auth_headers=mcs_auth_headers,
                 )
 
         # Check for cancellation before starting
@@ -345,6 +353,28 @@ async def _get_cdr_url(job_id: int) -> str:
     return settings.DEFAULT_CDR_URL
 
 
+async def _get_mcs_auth_headers(job_id: int) -> dict[str, str]:
+    """Resolve auth headers by reading live credentials from the referenced MCS config.
+
+    Mirrors `_get_cdr_auth_headers`: the job snapshots `mcs_id`, and credentials are
+    read from the live config rather than duplicated onto the job row, so secrets
+    live in exactly one place.
+
+    A job with no `mcs_id` predates MCS connections or targets the env-var engine —
+    unauthenticated, so no headers. A job whose linked config has been deleted has
+    unrecoverable credentials and raises rather than silently evaluating without
+    auth, which a remote MCS would reject with 401 on every patient.
+    """
+    async with async_session() as session:
+        job = await session.get(Job, job_id)
+        if job is None or job.mcs_id is None:
+            return {}
+        cfg = await session.get(MCSConfig, job.mcs_id)
+        if cfg is None:
+            raise RuntimeError(f"MCS config {job.mcs_id} referenced by job {job_id} no longer exists.")
+        return await _build_auth_headers(cfg.auth_type, cfg.auth_credentials)
+
+
 async def _get_mcs_url(job_id: int) -> str:
     """Resolve the MCS URL for a job.
 
@@ -370,6 +400,7 @@ async def _process_single_batch(
     cdr_url: str,
     auth_headers: dict[str, str],
     mcs_url: str,
+    mcs_auth_headers: dict[str, str] | None = None,
 ) -> None:
     """Process a single batch in two phases.
 
@@ -431,7 +462,11 @@ async def _process_single_batch(
                 try:
                     gather_result = await strategy.gather_patient_data(cdr_url, patient_id, auth_headers)
                     if gather_result.resources:
-                        await push_resources(gather_result.resources)
+                        await push_resources(
+                            gather_result.resources,
+                            target_url=mcs_url,
+                            auth_headers=mcs_auth_headers,
+                        )
                     logger.info(
                         f"Pushed {len(gather_result.resources)} resources for {patient_id[:8]}",
                         extra={"job_id": job_id, "patient_id": patient_id},
@@ -546,7 +581,12 @@ async def _process_single_batch(
 
                 try:
                     measure_report = await evaluate_measure(
-                        measure_id, patient_id, period_start, period_end, measure_engine_url=mcs_url
+                        measure_id,
+                        patient_id,
+                        period_start,
+                        period_end,
+                        measure_engine_url=mcs_url,
+                        auth_headers=mcs_auth_headers,
                     )
 
                     populations = _extract_populations(measure_report)
@@ -564,7 +604,7 @@ async def _process_single_batch(
                     # new rows without refs.
                     evaluated_resources_snapshot: list[dict] | None = None
                     try:
-                        snapshot_result = await snapshot_evaluated_resources(measure_report)
+                        snapshot_result = await snapshot_evaluated_resources(measure_report, mcs_url, mcs_auth_headers)
                         evaluated_resources_snapshot = snapshot_result if snapshot_result is not None else []
                     except Exception as snap_exc:
                         logger.warning(

backend/tests/test_services_fhir_client.py

@@ -1032,7 +1032,7 @@ async def test_snapshot_evaluated_resources_returns_resolved_list():
         "Encounter/e1": {"resourceType": "Encounter", "id": "e1"},
     }
 
-    async def fake_resolve(ref):
+    async def fake_resolve(ref, base_url=None, auth_headers=None):
         return fake_resources[ref]
 
     with patch("app.services.fhir_client.resolve_evaluated_resource", side_effect=fake_resolve):
@@ -1054,7 +1054,7 @@ async def test_snapshot_evaluated_resources_skips_failed_refs():
         ],
     }
 
-    async def fake_resolve(ref):
+    async def fake_resolve(ref, base_url=None, auth_headers=None):
         if ref == "Encounter/e1":
             raise RuntimeError("404 not found")
         return {"resourceType": ref.split("/")[0], "id": ref.split("/")[1]}
@@ -1852,3 +1852,69 @@ def test_wait_for_valueset_expansion_logs_timeout(monkeypatch, caplog):
 
     assert expanded == {}
     assert "ValueSet expansion timed out" in caplog.text
+
+
+# ---------------------------------------------------------------------------
+# MCS auth wiring — evaluate_measure / resolve_evaluated_resource
+# (regression: remote MCS connections got 401 because no auth was ever sent)
+# ---------------------------------------------------------------------------
+
+
+async def test_evaluate_measure_sends_auth_headers(mock_measure_report):
+    """evaluate_measure forwards auth headers to the MCS."""
+    mock_response = _make_response(200, mock_measure_report)
+
+    with patch("app.services.fhir_client.httpx.AsyncClient") as mock_httpx:
+        mock_ctx = AsyncMock()
+        mock_ctx.get = AsyncMock(return_value=mock_response)
+        mock_httpx.return_value.__aenter__ = AsyncMock(return_value=mock_ctx)
+        mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False)
+
+        await evaluate_measure(
+            "measure-1",
+            "patient-1",
+            "2024-01-01",
+            "2024-12-31",
+            measure_engine_url="https://mcs.example.org/fhir",
+            auth_headers={"Authorization": "Bearer tok-123"},
+        )
+
+    sent = mock_ctx.get.call_args.kwargs.get("headers")
+    assert sent is not None, "evaluate_measure must pass headers to the MCS"
+    assert sent.get("Authorization") == "Bearer tok-123"
+
+
+async def test_evaluate_measure_without_auth_sends_no_authorization(mock_measure_report):
+    """Unauthenticated MCS (local HAPI) keeps working — no Authorization header."""
+    mock_response = _make_response(200, mock_measure_report)
+
+    with patch("app.services.fhir_client.httpx.AsyncClient") as mock_httpx:
+        mock_ctx = AsyncMock()
+        mock_ctx.get = AsyncMock(return_value=mock_response)
+        mock_httpx.return_value.__aenter__ = AsyncMock(return_value=mock_ctx)
+        mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False)
+
+        await evaluate_measure("measure-1", "patient-1", "2024-01-01", "2024-12-31")
+
+    sent = mock_ctx.get.call_args.kwargs.get("headers") or {}
+    assert "Authorization" not in sent
+
+
+async def test_resolve_evaluated_resource_uses_given_base_and_auth():
+    """Snapshot reads target the job's MCS with its credentials, not the env default."""
+    mock_response = _make_response(200, {"resourceType": "Condition", "id": "c1"})
+
+    with patch("app.services.fhir_client.httpx.AsyncClient") as mock_httpx:
+        mock_ctx = AsyncMock()
+        mock_ctx.get = AsyncMock(return_value=mock_response)
+        mock_httpx.return_value.__aenter__ = AsyncMock(return_value=mock_ctx)
+        mock_httpx.return_value.__aexit__ = AsyncMock(return_value=False)
+
+        await resolve_evaluated_resource(
+            "Condition/c1",
+            base_url="https://mcs.example.org/fhir",
+            auth_headers={"Authorization": "Bearer tok-123"},
+        )
+
+    assert mock_ctx.get.call_args[0][0] == "https://mcs.example.org/fhir/Condition/c1"
+    assert mock_ctx.get.call_args.kwargs.get("headers", {}).get("Authorization") == "Bearer tok-123"