feat: FHIR R4 conformance, integration push resilience, OCR + worker fixes

Squash merge of fix/fhir-r4-conformance (49 commits).

FHIR R4 facade conformance & honesty (F2-F18):
- Advertise R4 4.0.1 (was R4B 4.3.0); honest CapabilityStatement
  (drop unimplemented batch/transaction/history/validate, dynamic date)
- Honest versioning (no-version) + If-Match optimistic locking (412)
- Spec-compliant pagination links (page=N), _elements projection,
  _summary=count, _total=none, _format=xml explicit reject
- Token/category search via JSONB @> containment; date precision
  semantics; sort column corrections; search.mode on every entry

FHIR import pipeline (G6-G11, I4-I7):
- Import all 15 resource types (was 7); honor entry.request.method
  (PUT/POST/DELETE/ifNoneExist) for transaction bundles
- Record Provenance per entry; cross-tenant collision warnings
- Reference routing for bare urn:uuid via FIELD_HINT_TO_TYPE
- Observation.component + canonical interpretation JSONB columns
- validate_and_filter_observations no longer mutates caller's list
- Patient schema 0..* cardinality for name/address/telecom

Tenant bridge & auth (F19, G10):
- Service-account JWT (POST /auth/service-account, password-less)
- X-Tenant header override for SYSTEM_ADMIN
- Provenance.agent.who resolves to real Practitioner/Device (F12)
- DocumentReference.author resolves to Practitioner (F11)

Integration sync & push (H1-H8):
- Shared run_sync pipeline (worker + manual endpoint)
- Push resilience: per-row 401 retry, cursor integrity, batch isolation
- Remote Provenance POST after push; insufficient_scope detection
- OAuth hygiene: atomic state consume, 429 classification, token
  revocation (RFC 7009), DCR guard, OperationOutcome diagnostics

Worker & notification stability (A7, C11-C13):
- Worker-scoped NullPool async engine singleton (fixes loop-affinity crash)
- Inject worker session into notification trigger task
- Deactivate dead push subscriptions on HTTP 410/404
- Skip LLM call when all OCR docs produce empty text

OCR pipeline fixes (C7-C9):
- Compute relative_score in OCR path (was always NULL)
- Correct normalized_value direction (raw -> preferred unit)
- Case-insensitive exact match for catalog dedup (was ilike wildcard bug)

Hygiene & config (B11, B14, B15, J9):
- VAPID keys required in production (prod-guard validator)
- setup_env.py auto-generates VAPID key pair + prompts contact email
- check_observation_access defensive against missing reference key
- Remove print()/traceback wrapper from doctors endpoint
- Lifespan startup fail-fatal in production

Security: redact Web Push endpoints in logs; use fake Fernet key in tests.

DB migrations required: f11a2b3c4d5e, f19a2b3c4d5e, i6a7b8c9d0e1, i7b8c9d0e1f2

Author: constLiakos

.env.example

@@ -92,8 +92,11 @@ JWT_EXPIRATION_HOURS=24
 # SMTP_PASSWORD=your_app_password
 # SMTP_FROM=noreply@health-assistant.local
 
-# ─── Web Push (VAPID) — required for browser notifications ────────────────
-# Generate with: npx web-push generate-vapid-keys
+# ─── Web Push (VAPID) — required in production for browser notifications ───
+# REQUIRED in production — boot refuses to start without these when
+# APP_ENV != "development" (see config.py prod-guard).
+# Generate either with: scripts/setup_env.py (auto-fills these), OR manually
+# with: npx web-push generate-vapid-keys
 # VAPID_PUBLIC_KEY=
 # VAPID_PRIVATE_KEY=
 # VAPID_ADMIN_EMAIL=admin@example.com

CHANGELOG.md

@@ -0,0 +1,88 @@
+"""Add practitioner_id to documents for FHIR DocumentReference.author (F11)
+
+Revision ID: f11a2b3c4d5e
+Revises: d2e3f4a5b6c7
+Create Date: 2026-06-30 00:00:00.000000
+
+Closes audit F11: ``DocumentReference.author`` previously emitted
+``Practitioner/<owner_id>`` but ``documents.owner_id`` is a ``ForeignKey``
+to ``users.id``, not ``doctors.id`` — so external clients resolving the
+reference got 404. This migration adds a nullable ``practitioner_id``
+column (FK to ``doctors.id``, ON DELETE SET NULL), backfills it from the
+existing owner→doctor mapping, and adds an index for search.
+
+The application layer (``DocumentModel.to_fhir_dict``) emits
+``Practitioner/<practitioner_id>`` when the column is set, and omits the
+``author`` element otherwise (rather than emitting a wrong reference).
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+revision: str = "f11a2b3c4d5e"
+down_revision: Union[str, Sequence[str], None] = "d2e3f4a5b6c7"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # 1. Add the nullable column. Type matches doctors.id (UUID).
+    op.add_column(
+        "documents",
+        sa.Column(
+            "practitioner_id",
+            postgresql.UUID(as_uuid=True),
+            nullable=True,
+            comment="Resolved Practitioner (DoctorModel) id for FHIR "
+            "DocumentReference.author. Backfilled from owner_id at migration "
+            "time; set on new uploads via owner→doctor lookup.",
+        ),
+    )
+
+    # 2. Backfill from the existing owner→doctor mapping. Each user may have
+    #    at most one doctor row per tenant; pick the first match deterministically.
+    op.execute(
+        """
+        UPDATE documents d
+        SET practitioner_id = sub.id
+        FROM (
+            SELECT DISTINCT ON (d2.tenant_id, d2.owner_id)
+                   d2.id, d2.tenant_id, d2.owner_id
+            FROM documents d2
+            JOIN doctors doc ON doc.user_id = d2.owner_id
+                            AND doc.tenant_id = d2.tenant_id
+            WHERE d2.owner_id IS NOT NULL
+            ORDER BY d2.tenant_id, d2.owner_id, doc.id
+        ) AS sub
+        WHERE d.tenant_id = sub.tenant_id
+          AND d.owner_id = sub.owner_id
+        """
+    )
+
+    # 3. Add the FK constraint + index.
+    op.create_foreign_key(
+        "fk_documents_practitioner_id_doctors",
+        "documents",
+        "doctors",
+        ["practitioner_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+    op.create_index(
+        "ix_documents_practitioner_id",
+        "documents",
+        ["practitioner_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_documents_practitioner_id", table_name="documents")
+    op.drop_constraint(
+        "fk_documents_practitioner_id_doctors",
+        "documents",
+        type_="foreignkey",
+    )
+    op.drop_column("documents", "practitioner_id")

backend/alembic/versions/f11a2b3c4d5e_add_practitioner_id_to_documents.py

@@ -0,0 +1,69 @@
+"""Service-account flag on users (F19)
+
+Revision ID: f19a2b3c4d5e
+Revises: i7b8c9d0e1f2
+Create Date: 2026-07-01
+
+Adds ``users.is_service_account`` (BOOLEAN NOT NULL DEFAULT false) and relaxes
+``users.hashed_password`` to nullable so password-less machine accounts can be
+created. Service accounts are minted by admins via ``POST /auth/service-account``
+and carry a long-lived JWT with ``is_service_account=True`` — they authenticate
+against the FHIR facade and REST API as a bearer token without a password.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "f19a2b3c4d5e"
+down_revision: Union[str, Sequence[str], None] = "i7b8c9d0e1f2"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "ALTER TABLE users "
+        "ADD COLUMN IF NOT EXISTS is_service_account BOOLEAN DEFAULT false"
+    )
+    op.execute(
+        "UPDATE users SET is_service_account = false WHERE is_service_account IS NULL"
+    )
+    op.alter_column(
+        "users", "is_service_account",
+        existing_type=sa.Boolean(),
+        nullable=False,
+        server_default="false",
+    )
+    # Relax hashed_password so service accounts (which have no password) can be created.
+    op.alter_column(
+        "users", "hashed_password",
+        existing_type=sa.String(255),
+        nullable=True,
+    )
+    op.create_index(
+        "ix_users_is_service_account",
+        "users",
+        ["is_service_account"],
+        postgresql_where=sa.text("is_service_account = true"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_users_is_service_account", table_name="users")
+    # Re-set NULL passwords to a dummy hash before re-tightening.
+    op.execute(
+        "UPDATE users SET hashed_password = '!' WHERE hashed_password IS NULL"
+    )
+    op.alter_column(
+        "users", "hashed_password",
+        existing_type=sa.String(255),
+        nullable=False,
+    )
+    op.alter_column(
+        "users", "is_service_account",
+        existing_type=sa.Boolean(),
+        nullable=True,
+        server_default=None,
+    )
+    op.execute("ALTER TABLE users DROP COLUMN IF EXISTS is_service_account")

backend/alembic/versions/f19a2b3c4d5e_service_account_flag.py

@@ -0,0 +1,59 @@
+"""Observation.interpretation: String -> JSONB (I6)
+
+Revision ID: i6a7b8c9d0e1
+Revises: f11a2b3c4d5e
+Create Date: 2026-06-30
+
+Stores the canonical FHIR R4 interpretation shape (``0..* CodeableConcept``,
+a list) instead of a flattened display string. The backfill wraps legacy
+plain-string rows in ``[{"text": <value>}]`` so the round-trip
+import → export preserves the LOINC/OBSINT coding instead of collapsing
+to a lossy display string.
+
+The reverse (downgrade) flattens the list back to a display string using
+the same precedence as ``_flatten_interpretation`` (coding.display →
+coding.code → text).
+"""
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "i6a7b8c9d0e1"
+down_revision: Union[str, Sequence[str], None] = "f11a2b3c4d5e"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE fhir_observations
+        ALTER COLUMN interpretation TYPE JSONB
+        USING CASE
+            WHEN interpretation IS NULL THEN NULL
+            WHEN interpretation ~ '^\\s*(\\[|\\{)' THEN interpretation::jsonb
+            ELSE jsonb_build_array(jsonb_build_object('text', interpretation))
+        END
+        """
+    )
+
+
+def downgrade() -> None:
+    op.execute(
+        """
+        ALTER TABLE fhir_observations
+        ALTER COLUMN interpretation TYPE TEXT
+        USING CASE
+            WHEN interpretation IS NULL THEN NULL
+            WHEN jsonb_typeof(interpretation) = 'array'
+                 AND jsonb_array_length(interpretation) > 0
+            THEN COALESCE(
+                interpretation->0->'coding'->0->>'display',
+                interpretation->0->'coding'->0->>'code',
+                interpretation->0->>'text',
+                ''
+            )
+            ELSE interpretation::text
+        END
+        """
+    )

backend/alembic/versions/i6a7b8c9d0e1_observation_interpretation_jsonb.py

@@ -0,0 +1,42 @@
+"""Observation.component JSONB column (prereq for H2)
+
+Revision ID: i7b8c9d0e1f2
+Revises: i6a7b8c9d0e1
+Create Date: 2026-06-30
+
+Adds the canonical FHIR R4 ``Observation.component`` (``0..*``) column as
+JSONB so multi-component observations (blood pressure, panels like BMP/CBC)
+can be stored and round-tripped. Without this column, the push path (H2)
+cannot emit ``component[]`` even after the SDK mapper is fixed, because push
+reads ``obs.to_fhir_dict()`` which projects from ORM columns.
+
+A GIN index supports future ``component-code-value-*`` search params.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = "i7b8c9d0e1f2"
+down_revision: Union[str, Sequence[str], None] = "i6a7b8c9d0e1"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "fhir_observations",
+        sa.Column("component", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+    )
+    op.create_index(
+        "ix_fhir_observations_component_gin",
+        "fhir_observations",
+        ["component"],
+        postgresql_using="gin",
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_fhir_observations_component_gin", table_name="fhir_observations")
+    op.drop_column("fhir_observations", "component")

backend/alembic/versions/i7b8c9d0e1f2_observation_component_jsonb.py

@@ -10,7 +10,7 @@
 import logging
 from typing import Any, Dict
 
-from sqlalchemy import select
+from sqlalchemy import select, func
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.biomarker_model import BiomarkerDefinition, Unit
@@ -81,7 +81,7 @@ async def process_unknown_medications(
         name_map[def_data.raw_name_match] = def_data.name
         existing = await db.execute(
             select(MedicationCatalog).where(
-                MedicationCatalog.name.ilike(def_data.name)
+                func.lower(MedicationCatalog.name) == func.lower(def_data.name)
             )
         )
         if not existing.scalar_one_or_none():

backend/app/ai/pipeline/ontology.py

@@ -29,7 +29,7 @@
 from app.models.fhir.medication import Medication, MedicationCatalog, MedicationStatus
 from app.models.fhir.patient import Observation
 from app.models.document_model import DocumentModel
-from app.services.fhir_helpers import FhirSerializationError, assert_valid_fhir
+from app.services.fhir_helpers import FhirSerializationError, _normalize_interpretation, assert_valid_fhir
 
 logger = logging.getLogger(__name__)
 
@@ -196,19 +196,55 @@ async def save_observation(
     biomarker_id = target_bio.id if target_bio else None
     unit_symbol = b.unit_symbol
     raw_unit_id = None
+    # Default normalized_value to the raw value; refined below if we have
+    # enough unit information to convert. The previous code only normalized
+    # when the biomarker had a preferred unit *different* from the matched
+    # unit, leaving auto-created biomarkers (no preferred unit) with mixed-
+    # unit trends across labs. Now we always normalize at least to the
+    # base SI unit so trends are consistent.
     normalized_val = val_float
 
+    # --- relative_score ---------------------------------------------------
+    # Mirror ObservationBuilder.build() (integrations path): the value's
+    # position within the reference range as a [0.0, 1.0] float. The
+    # previous OCR path never set relative_score, leaving the biomarker
+    # engine's flagship scoring feature silently disabled for OCR data.
+    relative_score: Optional[float] = None
+    ref_min = b.reference_range_min
+    ref_max = b.reference_range_max
+    if val_float is not None and ref_min is not None and ref_max is not None and ref_max > ref_min:
+        relative_score = (val_float - ref_min) / (ref_max - ref_min)
+        relative_score = max(0.0, min(1.0, relative_score))
+    elif ref_min is not None or ref_max is not None:
+        # Incomplete range — middle score, matches ObservationBuilder.
+        relative_score = 0.5
+
     if unit_symbol:
         unit_lower = unit_symbol.lower()
         if unit_lower in units_by_symbol:
             matched_unit = units_by_symbol[unit_lower]
             raw_unit_id = matched_unit.id
-            if (
-                target_bio
-                and target_bio.preferred_unit_id
-                and str(target_bio.preferred_unit_id) != str(matched_unit.id)
-            ):
-                normalized_val = val_float * matched_unit.conversion_multiplier
+            # Convert raw value -> base SI unit. matched_unit.conversion_multiplier
+            # is "multiply this unit's value by this number to get the base unit".
+            base_value = val_float * matched_unit.conversion_multiplier
+            # Express normalized_value in the biomarker's preferred unit
+            # (fall back to base SI when no preferred unit is set, so trends
+            # are at least consistent across labs reporting in different raw
+            # units). The previous code skipped normalization entirely when
+            # no preferred_unit_id was set, AND used the wrong direction
+            # (raw * raw_mult = base, then labeled it as-preferred).
+            preferred_unit: Optional[Unit] = None
+            if target_bio and target_bio.preferred_unit_id:
+                # Look up the preferred Unit object. units_by_symbol is the
+                # only Unit collection passed in; scan it once (small set).
+                for u in units_by_symbol.values():
+                    if str(u.id) == str(target_bio.preferred_unit_id):
+                        preferred_unit = u
+                        break
+            if preferred_unit is not None and preferred_unit.conversion_multiplier:
+                normalized_val = base_value / preferred_unit.conversion_multiplier
+            else:
+                normalized_val = base_value
         else:
             new_unit = Unit(
                 symbol=unit_symbol,
@@ -247,9 +283,10 @@ async def save_observation(
         raw_value=val_float,
         raw_unit_id=raw_unit_id,
         normalized_value=normalized_val,
+        relative_score=relative_score,
         lab_reference_range=lab_ref_range,
         method=b.method,
-        interpretation=b.interpretation_flag,
+        interpretation=_normalize_interpretation(b.interpretation_flag),
         category=_fhir_observation_category(
             target_bio.category if target_bio else None
         ),