diff --git a/docs/ELIGIBILITY_API_WATERFALL.md b/docs/ELIGIBILITY_API_WATERFALL.md new file mode 100644 index 00000000..2a2dd666 --- /dev/null +++ b/docs/ELIGIBILITY_API_WATERFALL.md @@ -0,0 +1,167 @@ +# Governed API-first eligibility + +OpenAdapt uses a sanctioned 270/271 API when an exact practice/payer route is +available, then uses compiled portal replay only as a reviewed fallback. Every +answer is qualified by service, date, network, coverage level, and time period; +ambiguous responses enter the attended queue. + +## Execution contract + +```text +exact practice payer binding + ├─ API + │ ├─ exact, unambiguous 271 → atomic local evidence → verify → complete + │ ├─ transient failure → bounded retry → reviewed portal or queue + │ └─ identity/config/data ambiguity → attended queue + ├─ reviewed portal route → compiled replay + independent effect check + ├─ queue → practice staff resolves with evidence + └─ excluded → no automated action +``` + +Unknown payer names, unmatched payer IDs, unreviewed service types, and account +or test/production mode mismatches never select an API or portal implicitly. + +The public `payer_routes.yaml` is one synthetic Stedi TEST-mode example. +Production payer maps are practice-scoped deployment data. Each API binding +records: + +- the exact request payer ID, including leading zeroes; +- Stedi's stable five-character payer ID; +- a digest of the reviewed payer-directory record; +- supported service-type codes; +- the practice account and application mode; +- the verification date and an explicit reviewed-portal-fallback decision. + +This keeps the route mechanism open while keeping productionized payer recipes +inside the deployment boundary. + +The committed synthetic Cigna binding was rechecked against Stedi's public +Payer Network on 2026-07-21: primary payer ID `62308`, current Stedi payer ID +`HGJLR` (`SX071` is an alias), and dental eligibility support. Both exact IDs +resolve to the same reviewed route; fuzzy payer search never selects a route. + +## Parsing without “first match” shortcuts + +`parse_271` retains every qualified patient-responsibility entry and populates +practice-facing convenience values only when one value matches the request's: + +- service type or procedure; +- in-network, out-of-network, or not-applicable qualifier; +- individual/family coverage level; +- benefit date and requested date of service; +- time qualifier, including separate total (`23`, calendar year) and remaining + (`29`) deductible/out-of-pocket amounts. + +Conflicting active/inactive signals, a response for another service, a payer or +application-mode mismatch, an active response without an explicit coverage +interval containing the requested service date, or two different values with +the same qualifiers is not an answer. Redirects and every other non-2xx HTTP +status are also never interpreted as successful eligibility responses. + +The current request schema is deliberately subscriber-only and requires the +member ID. The returned 271 must contain the same subscriber member ID; every +name or birth date supplied in the request must also be present and match after +conservative Unicode, case, and whitespace normalization. The returned provider +NPI must also exactly match the request. A missing subscriber or provider, a +mismatched identity, or a response containing dependent subjects enters the +attended queue. A dependent response cannot satisfy this subscriber request +contract; it requires an explicitly typed dependent request rather than a guess +about which returned person is the patient. + +## Error and fallback rules + +Only explicitly transient outcomes retry automatically: + +- HTTP `429`; +- HTTP `5xx` and network timeouts/failures; +- payer connectivity AAA `42`, `80`, or Stedi's documented `42` + `79` + combination. + +Retries are bounded. After they are exhausted, the result may use a portal +fallback only when that fallback was explicitly reviewed and the portal is not +barred. Authentication, invalid payer or +request, provider configuration, member identity, and ambiguous response +outcomes go to practice staff without automatic retry or portal substitution. + +A 270 is a read, so retrying does not duplicate a business write. The local +`operation_id` binds attempts and artifact promotion to one logical check. + +## Practice-held account and PHI boundary + +The practice owns the Stedi account and credential. Production mode requires an +explicit practice-held-account and BAA confirmation. The API key is resolved +from an environment reference and never enters a result, artifact, or reason +string. Requests and response bodies are never logged. + +PHI-bearing evidence is written only through `PracticeArtifactPolicy`, which +requires: + +- an explicit local PHI boundary; +- an explicit TEST or PRODUCTION application mode that cannot be mixed within + one artifact boundary; +- either attested encrypted-volume storage or application AES-256-GCM; +- owner-only artifact, transaction, and file paths that are rechecked on every + reuse and index rebuild; +- a retention period and `egress: none`; +- symlink-safe, exclusive writes and a bounded single-writer lock. + +Only an exact, unambiguous answer can enter the consumable result store. The +artifact writer requires the original `EligibilityRequest`, verifies its +canonical SHA-256 and the response-subject evidence digest, then reparses the +exact raw response and requires canonical semantic equality with the supplied +normalized result. The policy's application mode—not a caller-selected result +field—is authoritative for that reparse and is bound into `boundary.json`, the +transaction manifest, normalized JSON, and derived CSV. A TEST response cannot +enter a PRODUCTION artifact boundary. It derives member/date/benefit-selector +fields from the request rather than accepting caller-supplied identity labels. The subject +evidence digest binds the request digest, raw-response digest, and verified +subscriber role; it does not hash a standalone low-entropy identity tuple or add +extra plaintext identity fields to the manifest. +The raw response and normalized practice record are staged, hashed, fsynced, +re-read, and verified together before either the transaction directory or its +prospective derived CSV index is promoted. A failed promotion or independent +hash-effect verification rolls the new transaction back. The CSV is a derived +index, not the source of truth. Repeating the same `operation_id` and content is +idempotent; reusing it with different content fails. Spreadsheet formula +prefixes are neutralized. The persisted `committed_at` is generated by the +writer; a caller-supplied observation timestamp is not accepted as audit truth. +Hash effects independently re-read both stored files. + +Retention expiry is bound to the policy and commit time. Every write refuses an +invalid retention manifest and purges expired transactions before idempotency +or index reuse. Deployments also call +`purge_expired_eligibility_artifacts(...)` on startup and their retention +schedule; it verifies the prospective index before atomically hiding expired +transactions, then deletes them. Malformed, insecure, or policy-mismatched +artifacts deny cleanup instead of deleting an unverified path. + +## Qualification + +The deterministic contract suite covers routing, qualifiers, conflicting +benefits, the HTTP/AAA taxonomy, retries, PHI-safe diagnostics, atomicity, +idempotency, encryption, tampering, symlink substitution, and package import +behavior. + +The live test runs Stedi's published synthetic Cigna dental request three times: + +```bash +export STEDI_API_KEY='' +pytest -q tests/test_eligibility_live_stedi.py -rs +``` + +All three trials must return an exact TEST-mode answer and independently verify +the encrypted raw and normalized artifacts. If the key is absent, the tests +skip and explicitly state that no live evidence was collected. + +## Primary references + +- [Real-Time Eligibility Check JSON](https://www.stedi.com/docs/healthcare/api-reference/post-healthcare-eligibility) +- [Submit eligibility checks](https://www.stedi.com/docs/healthcare/send-eligibility-checks) +- [Eligibility troubleshooting and retries](https://www.stedi.com/docs/healthcare/eligibility-troubleshooting) +- [Patient responsibility qualifiers](https://www.stedi.com/docs/healthcare/eligibility-patient-responsibility-benefits) +- [Payer retrieval](https://www.stedi.com/docs/healthcare/api-reference/get-payer) +- [Integrated practice accounts](https://www.stedi.com/docs/healthcare/integrated-account-overview) +- [Account, test keys, production, and BAA setup](https://www.stedi.com/docs/healthcare/account-settings) + +The client is import-light and is not imported by the recorder, compiler, or +replayer unless the eligibility waterfall is used. diff --git a/openadapt_flow/eligibility/__init__.py b/openadapt_flow/eligibility/__init__.py new file mode 100644 index 00000000..a363b130 --- /dev/null +++ b/openadapt_flow/eligibility/__init__.py @@ -0,0 +1,116 @@ +"""Governed API-first 270/271 eligibility mechanism. + +The client, exact route-binding schema, qualifier-aware parser, and atomic +practice-local artifact boundary are public. Practice payer maps and account +credentials are deployment data. Transient API failures can use a reviewed +portal fallback; identity, configuration, ambiguity, and forbidden-portal +outcomes enter the attended queue instead of being guessed. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover + from openadapt_flow.eligibility.artifact import ( + ArtifactEncryption, + EligibilityArtifact, + PracticeArtifactPolicy, + purge_expired_eligibility_artifacts, + write_and_verify, + write_eligibility_artifacts, + ) + from openadapt_flow.eligibility.client import ( + ApplicationMode, + BenefitSelection, + EligibilityRequest, + EligibilityResult, + EligibilityStatus, + ErrorCategory, + QualifiedBenefit, + RetryDisposition, + StediAccountBoundary, + StediEligibilityClient, + eligibility_request_sha256, + ) + from openadapt_flow.eligibility.waterfall import ( + EligibilityRoute, + PayerCapability, + RouteDecision, + load_payer_routes, + resolve_route, + run_waterfall, + ) + +__all__ = [ + "ApplicationMode", + "ArtifactEncryption", + "BenefitSelection", + "EligibilityArtifact", + "EligibilityRequest", + "EligibilityResult", + "EligibilityRoute", + "EligibilityStatus", + "ErrorCategory", + "PayerCapability", + "PracticeArtifactPolicy", + "QualifiedBenefit", + "RetryDisposition", + "RouteDecision", + "StediAccountBoundary", + "StediEligibilityClient", + "eligibility_request_sha256", + "load_payer_routes", + "purge_expired_eligibility_artifacts", + "resolve_route", + "run_waterfall", + "write_and_verify", + "write_eligibility_artifacts", +] + +_CLIENT = ( + "ApplicationMode", + "BenefitSelection", + "EligibilityRequest", + "EligibilityResult", + "EligibilityStatus", + "ErrorCategory", + "QualifiedBenefit", + "RetryDisposition", + "StediAccountBoundary", + "StediEligibilityClient", + "eligibility_request_sha256", +) +_WATERFALL = ( + "EligibilityRoute", + "PayerCapability", + "RouteDecision", + "load_payer_routes", + "resolve_route", + "run_waterfall", +) +_ARTIFACT = ( + "ArtifactEncryption", + "EligibilityArtifact", + "PracticeArtifactPolicy", + "purge_expired_eligibility_artifacts", + "write_and_verify", + "write_eligibility_artifacts", +) + + +def __getattr__(name: str) -> object: + """Lazy re-exports (PEP 562) -- importing the package stays cheap.""" + if name in _CLIENT: + from openadapt_flow.eligibility import client + + return getattr(client, name) + if name in _WATERFALL: + from openadapt_flow.eligibility import waterfall + + return getattr(waterfall, name) + if name in _ARTIFACT: + from openadapt_flow.eligibility import artifact + + return getattr(artifact, name) + raise AttributeError(f"module {__name__!r} has no attribute {name!r}") diff --git a/openadapt_flow/eligibility/artifact.py b/openadapt_flow/eligibility/artifact.py new file mode 100644 index 00000000..e5acc9fe --- /dev/null +++ b/openadapt_flow/eligibility/artifact.py @@ -0,0 +1,1051 @@ +"""Atomic practice-local eligibility evidence artifacts. + +Each logical check is one atomically promoted transaction directory containing +the exact raw response, the exact practice-facing normalized record, and a +hash manifest. A CSV is a derived index, never the source of truth. Repeating +the same ``operation_id`` with identical content is idempotent; reusing it with +different content fails loud. +""" + +from __future__ import annotations + +import base64 +import csv +import hashlib +import io +import json +import os +import re +import shutil +import stat +from contextlib import contextmanager +from datetime import datetime, timedelta, timezone +from enum import Enum +from pathlib import Path +from typing import Iterator, Literal, Mapping, Optional, Union +from uuid import uuid4 + +from pydantic import BaseModel, Field, field_validator, model_validator + +from openadapt_flow.eligibility.client import ( + ApplicationMode, + EligibilityRequest, + EligibilityResult, + eligibility_request_sha256, + parse_271, +) +from openadapt_flow.runtime.effects.document_hash import DocumentHashVerifier +from openadapt_flow.runtime.effects.effect import ( + Effect, + EffectKind, + EffectVerdict, + ValueExpr, + Verdict, +) + +RESULTS_CSV = "eligibility_results.csv" +TRANSACTIONS_DIR = "transactions" +BOUNDARY_FILE = "boundary.json" +_ENCRYPTED_SUFFIX = ".enc" +_AES_PREFIX = b"OAE1" + +_CSV_COLUMNS = [ + "committed_at", + "application_mode", + "http_status", + "payer", + "payer_id", + "member_id", + "date_of_service", + "network_code", + "coverage_level_code", + "time_qualifier_code", + "procedure_code", + "status", + "plan_name", + "copay", + "coinsurance_percent", + "deductible_total", + "deductible_remaining", + "out_of_pocket_total", + "out_of_pocket_remaining", + "service_type_codes", + "source", + "raw_271_sha256", + "operation_id", + "request_sha256", +] + + +class ArtifactEncryption(str, Enum): + PLATFORM_VOLUME = "platform_volume" + APPLICATION_AES256_GCM = "application_aes256_gcm" + + +class PracticeArtifactPolicy(BaseModel): + """Explicit PHI storage, encryption, retention, and egress boundary.""" + + boundary_id: str + application_mode: ApplicationMode + encryption: ArtifactEncryption + retention_days: int = Field(ge=1, le=3650) + egress: Literal["none"] = "none" + phi_storage_allowed: bool = True + volume_encryption_attested: bool = False + encryption_key_env: Optional[str] = None + + @field_validator("boundary_id", "encryption_key_env") + @classmethod + def _safe_name(cls, value: Optional[str]) -> Optional[str]: + if value is not None and ( + not value + or len(value) > 128 + or any( + ch + not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._:-" + for ch in value + ) + ): + raise ValueError("boundary identifiers must be non-PHI operational names") + return value + + @model_validator(mode="after") + def _encryption_contract(self) -> "PracticeArtifactPolicy": + if not self.phi_storage_allowed: + raise ValueError( + "eligibility artifacts require an explicit PHI storage boundary" + ) + if self.encryption is ArtifactEncryption.PLATFORM_VOLUME: + if not self.volume_encryption_attested: + raise ValueError("platform-volume mode requires encryption attestation") + if self.encryption_key_env is not None: + raise ValueError("platform-volume mode does not use an application key") + elif not self.encryption_key_env: + raise ValueError( + "application encryption requires an environment key reference" + ) + return self + + +class EligibilityArtifact(BaseModel): + artifact_dir: str + transaction_dir: str + results_csv: str + raw_271_file: str + normalized_file: str + raw_271_sha256: str + normalized_sha256: str + application_mode: ApplicationMode + committed_at: str + created: bool + effects: list[Effect] = Field(default_factory=list) + + +def _canonical(value: object) -> bytes: + return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() + + +def _sha(payload: bytes) -> str: + return hashlib.sha256(payload).hexdigest() + + +def _ensure_secure_dir(path: Path) -> None: + if path.exists() or path.is_symlink(): + info = path.lstat() + if not stat.S_ISDIR(info.st_mode) or stat.S_ISLNK(info.st_mode): + raise ValueError("artifact root must be a regular directory, not a link") + if stat.S_IMODE(info.st_mode) != 0o700: + raise PermissionError( + "existing artifact directories must already be owner-only" + ) + else: + path.mkdir(parents=True, mode=0o700) + if stat.S_IMODE(path.lstat().st_mode) != 0o700: + raise PermissionError("artifact root is not owner-only") + + +def _require_secure_existing_dir(path: Path, *, label: str) -> None: + """Refuse a committed directory whose type or owner-only mode drifted.""" + try: + info = path.lstat() + except FileNotFoundError as exc: + raise ValueError(f"{label} directory is missing") from exc + if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode): + raise ValueError(f"{label} is not a regular directory") + if stat.S_IMODE(info.st_mode) & 0o077: + raise PermissionError(f"{label} directory must remain owner-only") + + +def _write_exclusive(path: Path, payload: bytes) -> None: + flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + fd = os.open(path, flags, 0o600) + try: + view = memoryview(payload) + while view: + written = os.write(fd, view) + if written <= 0: + raise OSError("short protected artifact write") + view = view[written:] + os.fsync(fd) + finally: + os.close(fd) + + +def _read_regular(path: Path) -> bytes: + flags = os.O_RDONLY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + fd = os.open(path, flags) + try: + opened = os.fstat(fd) + if not stat.S_ISREG(opened.st_mode): + raise ValueError("artifact is not a regular file") + if stat.S_IMODE(opened.st_mode) & 0o077: + raise PermissionError("artifact files must remain owner-only") + chunks: list[bytes] = [] + while chunk := os.read(fd, 1 << 20): + chunks.append(chunk) + return b"".join(chunks) + finally: + os.close(fd) + + +def _fsync_dir(path: Path) -> None: + if os.name == "nt": + return + fd = os.open(path, os.O_RDONLY) + try: + os.fsync(fd) + finally: + os.close(fd) + + +def _policy_payload(policy: PracticeArtifactPolicy) -> bytes: + return _canonical( + { + "schema_version": 2, + "boundary_id": policy.boundary_id, + "application_mode": policy.application_mode.value, + "encryption": policy.encryption.value, + "retention_days": policy.retention_days, + "egress": policy.egress, + "phi_storage_allowed": policy.phi_storage_allowed, + } + ) + + +def _bind_policy(root: Path, policy: PracticeArtifactPolicy) -> None: + expected = _policy_payload(policy) + marker = root / BOUNDARY_FILE + if marker.exists() or marker.is_symlink(): + if _read_regular(marker) != expected: + raise ValueError("artifact root is bound to a different PHI policy") + return + _write_exclusive(marker, expected) + _fsync_dir(root) + + +def _key( + policy: PracticeArtifactPolicy, env: Optional[Mapping[str, str]] +) -> Optional[bytes]: + if policy.encryption is ArtifactEncryption.PLATFORM_VOLUME: + return None + source = os.environ if env is None else env + encoded = source.get(policy.encryption_key_env or "", "") + try: + key = base64.urlsafe_b64decode(encoded.encode()) + except Exception as exc: # noqa: BLE001 - error is deliberately secret-free + raise ValueError("application encryption key is not valid base64") from exc + if len(key) != 32: + raise ValueError("application encryption key must decode to 32 bytes") + return key + + +def _protect(payload: bytes, *, key: Optional[bytes], aad: bytes) -> bytes: + if key is None: + return payload + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + nonce = os.urandom(12) + return _AES_PREFIX + nonce + AESGCM(key).encrypt(nonce, payload, aad) + + +def _unprotect(payload: bytes, *, key: Optional[bytes], aad: bytes) -> bytes: + if key is None: + return payload + if len(payload) < 16 or payload[:4] != _AES_PREFIX: + raise ValueError("encrypted artifact envelope is malformed") + from cryptography.hazmat.primitives.ciphers.aead import AESGCM + + return AESGCM(key).decrypt(payload[4:16], payload[16:], aad) + + +@contextmanager +def _artifact_lock(root: Path) -> Iterator[None]: + lock = root / ".eligibility-write.lock" + try: + lock.mkdir(mode=0o700) + except FileExistsError as exc: + raise BlockingIOError( + "another eligibility artifact writer holds the lock" + ) from exc + try: + yield + finally: + lock.rmdir() + + +def _transaction_id(policy: PracticeArtifactPolicy, operation_id: str) -> str: + return hashlib.sha256(f"{policy.boundary_id}:{operation_id}".encode()).hexdigest()[ + :24 + ] + + +def _normalized_payload( + result: EligibilityResult, + *, + request: EligibilityRequest, + committed_at: str, +) -> dict[str, object]: + selection = request.benefit_selection + return { + "schema_version": 3, + "operation_id": result.operation_id, + "committed_at": committed_at, + "application_mode": result.application_mode.value + if result.application_mode is not None + else "", + "http_status": result.http_status, + "payer": result.payer_name or "", + "payer_id": result.payer_id, + "member_id": request.member_id or "", + "date_of_service": request.date_of_service, + "network_code": selection.network_code or "", + "coverage_level_code": selection.coverage_level_code or "", + "time_qualifier_code": selection.time_qualifier_code or "", + "procedure_code": selection.procedure_code or "", + "status": result.status.value, + "plan_name": result.plan_name or "", + "plan_begin": result.plan_begin or "", + "plan_end": result.plan_end or "", + "coverage_by_service": { + k: v.value for k, v in result.coverage_by_service.items() + }, + "benefits": [benefit.model_dump(mode="json") for benefit in result.benefits], + "copay": result.copay or "", + "coinsurance_percent": result.coinsurance_percent or "", + "deductible_total": result.deductible_total or "", + "deductible_remaining": result.deductible_remaining or "", + "out_of_pocket_total": result.out_of_pocket_total or "", + "out_of_pocket_remaining": result.out_of_pocket_remaining or "", + "service_type_codes": list(result.service_type_codes), + "source": result.source, + "raw_271_sha256": result.raw_271_sha256, + "request_sha256": result.request_sha256, + "response_subject_sha256": result.response_subject_sha256, + } + + +def _semantic_result_payload(result: EligibilityResult) -> bytes: + """Canonical result meaning, excluding only observation-time metadata.""" + return _canonical( + result.model_dump( + mode="json", + exclude={"checked_at", "attempt_count", "raw_271", "raw_271_bytes"}, + ) + ) + + +def _safe_csv(value: object) -> str: + text = str(value if value is not None else "") + return "'" + text if text.startswith(("=", "+", "-", "@", "\t", "\r")) else text + + +def _csv_payload(rows: list[dict[str, object]]) -> bytes: + stream = io.StringIO(newline="") + writer = csv.DictWriter(stream, fieldnames=_CSV_COLUMNS) + writer.writeheader() + for row in rows: + csv_row: dict[str, str] = {} + for column in _CSV_COLUMNS: + value = row.get(column, "") + if isinstance(value, list): + value = " ".join(str(item) for item in value) + csv_row[column] = _safe_csv(value) + writer.writerow(csv_row) + return stream.getvalue().encode() + + +def _atomic_replace(path: Path, payload: bytes) -> None: + if path.is_symlink(): + raise ValueError("refusing to replace a symlinked artifact index") + temporary = path.with_name(f".{path.name}.{uuid4().hex}.tmp") + _write_exclusive(temporary, payload) + try: + os.replace(temporary, path) + _fsync_dir(path.parent) + finally: + if temporary.exists() or temporary.is_symlink(): + temporary.unlink() + + +def _effect(name: str, digest: str, probe: str) -> list[Effect]: + match = {"name": ValueExpr(literal=name)} + return [ + Effect( + kind=EffectKind.RECORD_WRITTEN, + match=match, + expected_count=1, + probe=f"exactly one {probe}", + ), + Effect( + kind=EffectKind.FIELD_EQUALS, + match=match, + field="sha256", + value=ValueExpr(literal=digest), + probe=f"{probe} storage bytes match the committed digest", + ), + ] + + +def _artifact_from_manifest( + root: Path, tx_dir: Path, manifest: dict[str, object], *, created: bool +) -> EligibilityArtifact: + raw_name = str(manifest["raw_file"]) + normalized_name = str(manifest["normalized_file"]) + effects = _effect(raw_name, str(manifest["raw_storage_sha256"]), "raw 271") + effects += _effect( + normalized_name, + str(manifest["normalized_storage_sha256"]), + "normalized eligibility record", + ) + return EligibilityArtifact( + artifact_dir=str(root), + transaction_dir=str(tx_dir), + results_csv=str(root / str(manifest["results_index"])), + raw_271_file=str(tx_dir / raw_name), + normalized_file=str(tx_dir / normalized_name), + raw_271_sha256=str(manifest["raw_plain_sha256"]), + normalized_sha256=str(manifest["normalized_plain_sha256"]), + application_mode=ApplicationMode(str(manifest["application_mode"])), + committed_at=str(manifest["committed_at"]), + created=created, + effects=effects, + ) + + +def _utc_timestamp(value: object, *, field: str) -> datetime: + try: + parsed = datetime.strptime(str(value), "%Y-%m-%dT%H:%M:%SZ").replace( + tzinfo=timezone.utc + ) + except ValueError as exc: + raise ValueError(f"eligibility transaction {field} is malformed") from exc + return parsed + + +def _load_manifest( + tx_dir: Path, + policy: Optional[PracticeArtifactPolicy] = None, + *, + transaction_id: Optional[str] = None, +) -> dict[str, object]: + raw = json.loads(_read_regular(tx_dir / "manifest.json")) + if not isinstance(raw, dict) or raw.get("schema_version") != 3: + raise ValueError("eligibility transaction manifest is malformed") + tx_id = transaction_id or tx_dir.name.removeprefix("tx_") + suffix = ( + _ENCRYPTED_SUFFIX + if policy is not None + and policy.encryption is ArtifactEncryption.APPLICATION_AES256_GCM + else "" + ) + expected = { + "raw_file": f"raw_271_{tx_id}.json{suffix}", + "normalized_file": f"result_{tx_id}.json{suffix}", + "results_index": RESULTS_CSV + suffix, + } + if not re.fullmatch(r"[0-9a-f]{24}", tx_id): + raise ValueError("eligibility transaction directory name is malformed") + if policy is not None and raw.get("boundary_id") != policy.boundary_id: + raise ValueError("eligibility transaction escaped its PHI boundary") + if ( + policy is not None + and raw.get("application_mode") != policy.application_mode.value + ): + raise ValueError("eligibility transaction application mode is malformed") + if not isinstance(raw.get("http_status"), int) or not ( + 200 <= int(raw["http_status"]) < 300 + ): + raise ValueError("eligibility transaction HTTP status is malformed") + committed_at = _utc_timestamp(raw.get("committed_at", ""), field="commit time") + retention_expires_at = _utc_timestamp( + raw.get("retention_expires_at", ""), field="retention expiry" + ) + if retention_expires_at <= committed_at: + raise ValueError("eligibility transaction retention interval is malformed") + if policy is not None and retention_expires_at != ( + committed_at + timedelta(days=policy.retention_days) + ): + raise ValueError( + "eligibility transaction retention expiry does not match its bound policy" + ) + for field, value in expected.items(): + if raw.get(field) != value: + raise ValueError(f"eligibility transaction {field} is malformed") + for field in ( + "raw_plain_sha256", + "raw_storage_sha256", + "normalized_plain_sha256", + "normalized_storage_sha256", + "request_sha256", + "response_subject_sha256", + ): + if not re.fullmatch(r"[0-9a-f]{64}", str(raw.get(field, ""))): + raise ValueError(f"eligibility transaction {field} is malformed") + if raw.get("egress") != "none": + raise ValueError("eligibility transaction egress policy is malformed") + return raw + + +def _verified_transaction_row( + tx_dir: Path, + policy: PracticeArtifactPolicy, + key: Optional[bytes], + *, + transaction_id: Optional[str] = None, + expected_manifest: Optional[dict[str, object]] = None, +) -> tuple[dict[str, object], dict[str, object]]: + """Re-read and verify every byte that defines one consumable result.""" + _require_secure_existing_dir(tx_dir, label="transaction") + manifest = _load_manifest(tx_dir, policy, transaction_id=transaction_id) + if expected_manifest is not None and manifest != expected_manifest: + raise ValueError("staged eligibility manifest does not match its contract") + final_tx_name = f"tx_{transaction_id}" if transaction_id else tx_dir.name + raw_name = str(manifest["raw_file"]) + raw_stored = _read_regular(tx_dir / raw_name) + if _sha(raw_stored) != manifest["raw_storage_sha256"]: + raise ValueError("raw artifact fails its committed storage digest") + raw_aad = f"{policy.boundary_id}:{final_tx_name}:{raw_name}".encode() + raw_plain = _unprotect(raw_stored, key=key, aad=raw_aad) + if _sha(raw_plain) != manifest["raw_plain_sha256"]: + raise ValueError("raw artifact fails its committed plaintext digest") + + normalized_name = str(manifest["normalized_file"]) + normalized_stored = _read_regular(tx_dir / normalized_name) + if _sha(normalized_stored) != manifest["normalized_storage_sha256"]: + raise ValueError("normalized artifact fails its committed storage digest") + normalized_aad = f"{policy.boundary_id}:{final_tx_name}:{normalized_name}".encode() + normalized_plain = _unprotect(normalized_stored, key=key, aad=normalized_aad) + if _sha(normalized_plain) != manifest["normalized_plain_sha256"]: + raise ValueError("normalized artifact fails its committed digest") + parsed = json.loads(normalized_plain) + if not isinstance(parsed, dict): + raise ValueError("normalized artifact is not an object") + expected_fields = { + "operation_id": manifest.get("operation_id"), + "request_sha256": manifest.get("request_sha256"), + "response_subject_sha256": manifest.get("response_subject_sha256"), + "application_mode": manifest.get("application_mode"), + "http_status": manifest.get("http_status"), + "committed_at": manifest.get("committed_at"), + "raw_271_sha256": manifest.get("raw_plain_sha256"), + } + for field, expected in expected_fields.items(): + if parsed.get(field) != expected: + raise ValueError(f"normalized artifact {field} does not match its manifest") + return manifest, parsed + + +def _index_material( + root: Path, + policy: PracticeArtifactPolicy, + key: Optional[bytes], + *, + now: datetime, + excluded: frozenset[Path] = frozenset(), + staged: Optional[tuple[Path, str, dict[str, object]]] = None, +) -> tuple[str, bytes, bytes]: + rows: list[dict[str, object]] = [] + transactions = root / TRANSACTIONS_DIR + for tx_dir in sorted(transactions.iterdir()): + if tx_dir in excluded or tx_dir.name.startswith("."): + continue + manifest, parsed = _verified_transaction_row(tx_dir, policy, key) + expires = _utc_timestamp( + manifest["retention_expires_at"], field="retention expiry" + ) + if expires <= now: + raise ValueError( + "expired eligibility transaction is not consumable; purge it first" + ) + rows.append(parsed) + if staged is not None: + stage, transaction_id, expected_manifest = staged + manifest, parsed = _verified_transaction_row( + stage, + policy, + key, + transaction_id=transaction_id, + expected_manifest=expected_manifest, + ) + expires = _utc_timestamp( + manifest["retention_expires_at"], field="retention expiry" + ) + if expires <= now: + raise ValueError("new eligibility transaction is already expired") + rows.append(parsed) + plain_csv = _csv_payload(rows) + name = RESULTS_CSV + (_ENCRYPTED_SUFFIX if key is not None else "") + payload = _protect( + plain_csv, + key=key, + aad=f"{policy.boundary_id}:index:{name}".encode(), + ) + return name, plain_csv, payload + + +def _stage_verified_index( + root: Path, + policy: PracticeArtifactPolicy, + key: Optional[bytes], + *, + name: str, + plain_csv: bytes, + payload: bytes, +) -> Path: + target = root / name + if target.is_symlink(): + raise ValueError("refusing to replace a symlinked artifact index") + if target.exists(): + _read_regular(target) + temporary = root / f".{name}.{uuid4().hex}.tmp" + try: + _write_exclusive(temporary, payload) + staged = _read_regular(temporary) + if staged != payload: + raise ValueError("staged eligibility index bytes changed during write") + observed_plain = _unprotect( + staged, + key=key, + aad=f"{policy.boundary_id}:index:{name}".encode(), + ) + if observed_plain != plain_csv: + raise ValueError("staged eligibility index fails plaintext verification") + return temporary + except Exception: + if temporary.exists() or temporary.is_symlink(): + temporary.unlink() + raise + + +def _promote_staged_index(temporary: Path, target: Path) -> None: + if target.is_symlink(): + raise ValueError("refusing to replace a symlinked artifact index") + os.replace(temporary, target) + _fsync_dir(target.parent) + + +def _restore_index(target: Path, previous: Optional[bytes]) -> None: + """Restore the pre-transaction index after a failed two-path promotion.""" + if previous is None: + if target.exists() or target.is_symlink(): + if target.is_symlink(): + raise ValueError("cannot roll back a symlinked artifact index") + target.unlink() + _fsync_dir(target.parent) + return + _atomic_replace(target, previous) + + +def _rebuild_index( + root: Path, + policy: PracticeArtifactPolicy, + key: Optional[bytes], + *, + now: Optional[datetime] = None, +) -> str: + effective_now = _effective_now(now) + name, plain_csv, payload = _index_material(root, policy, key, now=effective_now) + index_target = root / name + temporary = _stage_verified_index( + root, + policy, + key, + name=name, + plain_csv=plain_csv, + payload=payload, + ) + try: + _promote_staged_index(temporary, index_target) + finally: + if temporary.exists() and not temporary.is_symlink(): + temporary.unlink() + return name + + +def _effective_now(now: Optional[datetime]) -> datetime: + value = now or datetime.now(timezone.utc) + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("eligibility retention time must be timezone-aware") + return value.astimezone(timezone.utc) + + +def _purge_expired_locked( + root: Path, + policy: PracticeArtifactPolicy, + key: Optional[bytes], + *, + now: datetime, +) -> list[str]: + """Hide expired transactions, publish a verified index, then delete them.""" + transactions = root / TRANSACTIONS_DIR + expired: list[tuple[Path, str]] = [] + for tx_dir in sorted(transactions.iterdir()): + if tx_dir.name.startswith("."): + continue + _require_secure_existing_dir(tx_dir, label="transaction") + manifest = _load_manifest(tx_dir, policy) + expires = _utc_timestamp( + manifest["retention_expires_at"], field="retention expiry" + ) + if expires <= now: + expired.append((tx_dir, str(manifest["operation_id"]))) + if not expired: + return [] + + excluded = frozenset(path for path, _operation_id in expired) + name, plain_csv, payload = _index_material( + root, policy, key, now=now, excluded=excluded + ) + index_target = root / name + previous_index = ( + _read_regular(index_target) + if index_target.exists() or index_target.is_symlink() + else None + ) + temporary = _stage_verified_index( + root, + policy, + key, + name=name, + plain_csv=plain_csv, + payload=payload, + ) + quarantined: list[tuple[Path, Path]] = [] + try: + for original, _operation_id in expired: + quarantine = transactions / f".expired-{original.name}-{uuid4().hex}" + os.rename(original, quarantine) + quarantined.append((original, quarantine)) + _fsync_dir(transactions) + _promote_staged_index(temporary, index_target) + except Exception: + index_changed = not temporary.exists() + for original, quarantine in reversed(quarantined): + if quarantine.exists() and not quarantine.is_symlink(): + os.rename(quarantine, original) + _fsync_dir(transactions) + if index_changed: + _restore_index(index_target, previous_index) + raise + finally: + if temporary.exists() and not temporary.is_symlink(): + temporary.unlink() + + cleanup_errors: list[str] = [] + for _original, quarantine in quarantined: + try: + _require_secure_existing_dir(quarantine, label="expired transaction") + shutil.rmtree(quarantine) + except OSError as exc: + cleanup_errors.append(type(exc).__name__) + _fsync_dir(transactions) + if cleanup_errors: + raise RuntimeError( + "expired eligibility data was deactivated but secure deletion failed: " + + ",".join(cleanup_errors) + ) + return [operation_id for _path, operation_id in expired] + + +def purge_expired_eligibility_artifacts( + artifact_dir: Union[str, Path], + *, + policy: PracticeArtifactPolicy, + env: Optional[Mapping[str, str]] = None, + now: Optional[datetime] = None, +) -> list[str]: + """Purge expired PHI transactions and atomically remove them from the index. + + A malformed manifest, insecure path, missing encryption key, or failed + index verification denies the purge before any transaction is hidden. + """ + root = Path(artifact_dir) + if not root.exists() and not root.is_symlink(): + return [] + _ensure_secure_dir(root) + _bind_policy(root, policy) + key = _key(policy, env) + transactions = root / TRANSACTIONS_DIR + _ensure_secure_dir(transactions) + effective_now = _effective_now(now) + with _artifact_lock(root): + return _purge_expired_locked(root, policy, key, now=effective_now) + + +def write_eligibility_artifacts( + result: EligibilityResult, + artifact_dir: Union[str, Path], + *, + request: EligibilityRequest, + policy: PracticeArtifactPolicy, + env: Optional[Mapping[str, str]] = None, +) -> EligibilityArtifact: + """Atomically promote a PHI-bearing raw+normalized transaction.""" + if result.raw_271_bytes is None or result.raw_271_sha256 is None: + raise ValueError( + "a raw response is required for a consumable eligibility artifact" + ) + if result.application_mode is None or result.http_status is None: + raise ValueError("eligibility result lacks its response boundary metadata") + if not result.is_answer: + raise ValueError( + "only an exact unambiguous eligibility answer may be promoted as consumable" + ) + request_sha256 = eligibility_request_sha256(request) + if ( + result.operation_id != request.operation_id + or result.payer_id != request.payer_id + or result.service_type_codes != request.service_type_codes + or result.request_sha256 != request_sha256 + ): + raise ValueError("eligibility result is not bound to the supplied request") + if _sha(result.raw_271_bytes) != result.raw_271_sha256: + raise ValueError("raw eligibility bytes do not match the wire digest") + if result.application_mode is not policy.application_mode: + raise ValueError( + "eligibility result application mode does not match the artifact policy" + ) + reparsed = parse_271( + request, + result.raw_271_bytes, + http_status=result.http_status, + expected_mode=policy.application_mode, + ) + if not reparsed.is_answer: + raise ValueError("raw eligibility response is not a consumable answer") + if _semantic_result_payload(reparsed) != _semantic_result_payload(result): + raise ValueError( + "normalized eligibility result does not match the exact raw response" + ) + root = Path(artifact_dir) + _ensure_secure_dir(root) + _bind_policy(root, policy) + key = _key(policy, env) + effective_now = _effective_now(None) + transactions = root / TRANSACTIONS_DIR + _ensure_secure_dir(transactions) + tx_id = _transaction_id(policy, result.operation_id) + tx_dir = transactions / f"tx_{tx_id}" + raw_plain = result.raw_271_bytes + suffix = _ENCRYPTED_SUFFIX if key is not None else "" + raw_name = f"raw_271_{tx_id}.json{suffix}" + normalized_name = f"result_{tx_id}.json{suffix}" + + with _artifact_lock(root): + _purge_expired_locked(root, policy, key, now=effective_now) + if tx_dir.exists() or tx_dir.is_symlink(): + _require_secure_existing_dir(tx_dir, label="idempotency target") + manifest = _load_manifest(tx_dir, policy) + normalized_plain = _canonical( + _normalized_payload( + reparsed, + request=request, + committed_at=str(manifest["committed_at"]), + ) + ) + if ( + manifest.get("operation_id") != result.operation_id + or manifest.get("request_sha256") != request_sha256 + or manifest.get("raw_plain_sha256") != _sha(raw_plain) + or manifest.get("normalized_plain_sha256") != _sha(normalized_plain) + ): + raise FileExistsError( + "operation_id was already committed with different content" + ) + _rebuild_index(root, policy, key, now=effective_now) + return _artifact_from_manifest(root, tx_dir, manifest, created=False) + + stage = transactions / f".staging-{uuid4().hex}" + stage.mkdir(mode=0o700) + staged_index: Optional[Path] = None + index_target: Optional[Path] = None + previous_index: Optional[bytes] = None + promotion_started = False + tx_promoted = False + try: + committed_at = effective_now.strftime("%Y-%m-%dT%H:%M:%SZ") + normalized_plain = _canonical( + _normalized_payload( + reparsed, request=request, committed_at=committed_at + ) + ) + raw_aad = f"{policy.boundary_id}:{tx_dir.name}:{raw_name}".encode() + normalized_aad = ( + f"{policy.boundary_id}:{tx_dir.name}:{normalized_name}".encode() + ) + raw_stored = _protect(raw_plain, key=key, aad=raw_aad) + normalized_stored = _protect(normalized_plain, key=key, aad=normalized_aad) + _write_exclusive(stage / raw_name, raw_stored) + _write_exclusive(stage / normalized_name, normalized_stored) + expiry = (effective_now + timedelta(days=policy.retention_days)).strftime( + "%Y-%m-%dT%H:%M:%SZ" + ) + index_name = RESULTS_CSV + (_ENCRYPTED_SUFFIX if key is not None else "") + committed_manifest: dict[str, object] = { + "schema_version": 3, + "operation_id": result.operation_id, + "request_sha256": request_sha256, + "response_subject_sha256": result.response_subject_sha256, + "boundary_id": policy.boundary_id, + "application_mode": policy.application_mode.value, + "http_status": reparsed.http_status, + "committed_at": committed_at, + "retention_expires_at": expiry, + "egress": "none", + "raw_file": raw_name, + "raw_plain_sha256": _sha(raw_plain), + "raw_storage_sha256": _sha(raw_stored), + "normalized_file": normalized_name, + "normalized_plain_sha256": _sha(normalized_plain), + "normalized_storage_sha256": _sha(normalized_stored), + "results_index": index_name, + } + _write_exclusive(stage / "manifest.json", _canonical(committed_manifest)) + _fsync_dir(stage) + + _manifest, staged_row = _verified_transaction_row( + stage, + policy, + key, + transaction_id=tx_id, + expected_manifest=committed_manifest, + ) + if _canonical(staged_row) != normalized_plain: + raise ValueError( + "staged normalized eligibility record changed during verification" + ) + name, plain_csv, index_payload = _index_material( + root, + policy, + key, + now=effective_now, + staged=(stage, tx_id, committed_manifest), + ) + if name != index_name: + raise ValueError("staged eligibility index name changed") + staged_index = _stage_verified_index( + root, + policy, + key, + name=name, + plain_csv=plain_csv, + payload=index_payload, + ) + index_target = root / name + if index_target.exists() or index_target.is_symlink(): + previous_index = _read_regular(index_target) + + promotion_started = True + os.rename(stage, tx_dir) + tx_promoted = True + _fsync_dir(transactions) + _promote_staged_index(staged_index, index_target) + except Exception: + index_changed = staged_index is not None and not staged_index.exists() + if stage.exists() and not stage.is_symlink(): + shutil.rmtree(stage) + if tx_promoted and tx_dir.exists() and not tx_dir.is_symlink(): + _require_secure_existing_dir(tx_dir, label="failed transaction") + shutil.rmtree(tx_dir) + _fsync_dir(transactions) + if promotion_started and index_changed and index_target is not None: + _restore_index(index_target, previous_index) + raise + finally: + if staged_index is not None and staged_index.exists(): + if staged_index.is_symlink(): + raise ValueError("staged eligibility index became a symlink") + staged_index.unlink() + return _artifact_from_manifest(root, tx_dir, committed_manifest, created=True) + + +def _rollback_created_artifact( + artifact: EligibilityArtifact, + *, + policy: PracticeArtifactPolicy, + env: Optional[Mapping[str, str]], +) -> None: + """Remove a newly promoted transaction if independent verification fails.""" + root = Path(artifact.artifact_dir) + _ensure_secure_dir(root) + _bind_policy(root, policy) + key = _key(policy, env) + transactions = root / TRANSACTIONS_DIR + _require_secure_existing_dir(transactions, label="transactions") + tx_dir = Path(artifact.transaction_dir) + if tx_dir.parent != transactions or not re.fullmatch( + r"tx_[0-9a-f]{24}", tx_dir.name + ): + raise ValueError("refusing to roll back an unexpected transaction path") + effective_now = _effective_now(None) + with _artifact_lock(root): + if tx_dir.exists() or tx_dir.is_symlink(): + _require_secure_existing_dir(tx_dir, label="failed transaction") + shutil.rmtree(tx_dir) + _fsync_dir(transactions) + _purge_expired_locked(root, policy, key, now=effective_now) + _rebuild_index(root, policy, key, now=effective_now) + + +def write_and_verify( + result: EligibilityResult, + artifact_dir: Union[str, Path], + *, + request: EligibilityRequest, + policy: PracticeArtifactPolicy, + env: Optional[Mapping[str, str]] = None, +) -> tuple[EligibilityArtifact, list[EffectVerdict]]: + root = Path(artifact_dir) + artifact = write_eligibility_artifacts( + result, + root, + request=request, + policy=policy, + env=env, + ) + try: + # Re-open without following links before consulting the independent + # generic verifier. A newly created transaction is rolled back if any + # verification step fails, so a failed result never remains consumable. + _read_regular(Path(artifact.raw_271_file)) + _read_regular(Path(artifact.normalized_file)) + verifier = DocumentHashVerifier(root, glob=f"{TRANSACTIONS_DIR}/tx_*/*") + before = verifier.capture_pre_state() + verdicts = [verifier.verify(effect, before) for effect in artifact.effects] + if not all_confirmed(verdicts): + raise RuntimeError( + "eligibility artifact effect verification did not confirm" + ) + return artifact, verdicts + except Exception: + if artifact.created: + _rollback_created_artifact(artifact, policy=policy, env=env) + raise + + +def all_confirmed(verdicts: list[EffectVerdict]) -> bool: + return bool(verdicts) and all(v.verdict is Verdict.CONFIRMED for v in verdicts) diff --git a/openadapt_flow/eligibility/client.py b/openadapt_flow/eligibility/client.py new file mode 100644 index 00000000..c623eff4 --- /dev/null +++ b/openadapt_flow/eligibility/client.py @@ -0,0 +1,1270 @@ +"""Fail-closed Stedi 270/271 eligibility client. + +The public module is mechanism, not a production payer recipe. A deployment +supplies a practice-owned account boundary and an exact, reviewed payer binding +through :mod:`openadapt_flow.eligibility.waterfall`. + +Primary contract references (reviewed 2026-07-21): + +* https://www.stedi.com/docs/healthcare/api-reference/post-healthcare-eligibility +* https://www.stedi.com/docs/healthcare/send-eligibility-checks +* https://www.stedi.com/docs/healthcare/eligibility-troubleshooting +* https://www.stedi.com/docs/healthcare/eligibility-patient-responsibility-benefits +* https://www.stedi.com/docs/healthcare/integrated-account-overview + +No request body or response body is logged. Reasons contain only payer IDs, +HTTP/AAA codes, and fixed classifications. Raw 271 bytes are returned solely +for the explicit PHI-bearing local artifact boundary. +""" + +from __future__ import annotations + +import hashlib +import json +import re +import threading +import time +import unicodedata +from datetime import date, datetime, timezone +from enum import Enum +from typing import Any, Callable, Literal, Mapping, Optional +from uuid import uuid4 + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from openadapt_flow.runtime.effects.auth import AuthRef + +STEDI_ELIGIBILITY_URL = ( + "https://healthcare.us.stedi.com/2024-04-01/change/medicalnetwork/eligibility/v3" +) +SERVICE_TYPE_DENTAL = "35" +STEDI_API_KEY_ENV = "STEDI_API_KEY" +MAX_RESPONSE_BYTES = 8 * 1024 * 1024 + +_AAA_TRANSIENT = {"42", "80"} +_AAA_MEMBER = {"65", "67", "72", "73", "75", "76"} +_AAA_PROVIDER = { + "41", + "43", + "44", + "45", + "46", + "47", + "48", + "49", + "50", + "51", + "52", + "53", + "97", +} +_AAA_INVALID_PAYER = {"T4"} +_CODE_ACTIVE = "1" +_CODES_INACTIVE = {"6", "7", "8"} +_DATE_RE = re.compile(r"^\d{8}$") +_SAFE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") + + +class ApplicationMode(str, Enum): + TEST = "test" + PRODUCTION = "production" + + +class EligibilityStatus(str, Enum): + ACTIVE = "active" + INACTIVE = "inactive" + NOT_FOUND = "not_found" + PAYER_UNAVAILABLE = "payer_unavailable" + REJECTED = "rejected" + INDETERMINATE = "indeterminate" + + +class ErrorCategory(str, Enum): + AUTH_CONFIGURATION = "auth_configuration" + INVALID_PAYER = "invalid_payer" + INVALID_REQUEST = "invalid_request" + MEMBER_IDENTITY = "member_identity" + PROVIDER_CONFIGURATION = "provider_configuration" + THROTTLED = "throttled" + PAYER_TRANSIENT = "payer_transient" + TRANSPORT_TRANSIENT = "transport_transient" + SERVER_TRANSIENT = "server_transient" + RESPONSE_INVALID = "response_invalid" + RESPONSE_AMBIGUOUS = "response_ambiguous" + + +class RetryDisposition(str, Enum): + NO_RETRY_QUEUE = "no_retry_queue" + RETRY_THEN_PORTAL = "retry_then_portal" + RETRY_THEN_QUEUE = "retry_then_queue" + + +class StediAccountBoundary(BaseModel): + """Practice-held Stedi credential and tenancy boundary. + + Production keys are accepted only when the practice owns the account and + its BAA has been confirmed. The identifier must be operational, not PHI. + """ + + practice_account_id: str + application_mode: ApplicationMode + credential_env: str = STEDI_API_KEY_ENV + practice_holds_account: bool = True + baa_confirmed: bool = False + + @field_validator("practice_account_id", "credential_env") + @classmethod + def _safe_identifier(cls, value: str) -> str: + if not _SAFE_ID_RE.fullmatch(value): + raise ValueError("must be a non-PHI operational identifier") + return value + + @model_validator(mode="after") + def _production_boundary(self) -> "StediAccountBoundary": + if not self.practice_holds_account: + raise ValueError("Stedi credentials must belong to the practice account") + if ( + self.application_mode is ApplicationMode.PRODUCTION + and not self.baa_confirmed + ): + raise ValueError( + "production Stedi use requires the practice BAA confirmation" + ) + return self + + +class BenefitSelection(BaseModel): + """Qualifiers for the practice-facing normalized benefit values. + + The complete qualified benefit list is always retained. Convenience + values are populated only when this selector yields one unambiguous value. + """ + + network_code: Optional[Literal["Y", "N", "W"]] = None + coverage_level_code: Optional[str] = None + time_qualifier_code: Optional[str] = None + procedure_code: Optional[str] = None + + +class EligibilityRequest(BaseModel): + """One idempotently tracked 270 read. + + ``operation_id`` is local correlation/idempotency data and is never sent to + Stedi. PHI fields are suppressed from repr, and callers must not log the + result of ``model_dump``. + """ + + operation_id: str = Field(default_factory=lambda: str(uuid4())) + payer_id: str + member_id: str = Field(repr=False) + first_name: Optional[str] = Field(default=None, repr=False) + last_name: Optional[str] = Field(default=None, repr=False) + date_of_birth: Optional[str] = Field(default=None, repr=False) + provider_npi: str + provider_organization: Optional[str] = Field(default=None, repr=False) + provider_first_name: Optional[str] = Field(default=None, repr=False) + provider_last_name: Optional[str] = Field(default=None, repr=False) + service_type_codes: list[str] = Field(default_factory=lambda: [SERVICE_TYPE_DENTAL]) + date_of_service: str + benefit_selection: BenefitSelection = Field(default_factory=BenefitSelection) + + @field_validator("operation_id") + @classmethod + def _operation_id(cls, value: str) -> str: + if not _SAFE_ID_RE.fullmatch(value): + raise ValueError("operation_id must be an opaque non-PHI identifier") + return value + + @field_validator("payer_id") + @classmethod + def _payer_id(cls, value: str) -> str: + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,79}", value): + raise ValueError("payer_id must be an exact 1-80 character identifier") + return value + + @field_validator("member_id") + @classmethod + def _member_id(cls, value: str) -> str: + if not value or value != value.strip() or len(value) > 128: + raise ValueError("member_id is required for verified eligibility") + return value + + @field_validator("service_type_codes") + @classmethod + def _service_codes(cls, values: list[str]) -> list[str]: + if not values or len(values) > 99 or len(set(values)) != len(values): + raise ValueError("service_type_codes must contain 1-99 unique codes") + if any(not v or len(v) > 3 or v != v.strip() for v in values): + raise ValueError("service type codes must be exact non-empty strings") + return values + + @field_validator("date_of_birth", "date_of_service") + @classmethod + def _date8(cls, value: Optional[str]) -> Optional[str]: + if value is not None: + if not _DATE_RE.fullmatch(value): + raise ValueError("date must use YYYYMMDD") + datetime.strptime(value, "%Y%m%d") + return value + + @model_validator(mode="after") + def _identity_and_provider(self) -> "EligibilityRequest": + org = bool(self.provider_organization) + person = bool(self.provider_first_name and self.provider_last_name) + if org == person: + raise ValueError( + "provider requires exactly organization or first/last name" + ) + if not re.fullmatch(r"\d{10}", self.provider_npi): + raise ValueError("provider_npi must contain exactly 10 digits") + return self + + def to_stedi_body(self) -> dict[str, Any]: + provider: dict[str, Any] = {"npi": self.provider_npi} + if self.provider_organization: + provider["organizationName"] = self.provider_organization + else: + provider["firstName"] = self.provider_first_name + provider["lastName"] = self.provider_last_name + subscriber: dict[str, Any] = {} + for key, value in ( + ("firstName", self.first_name), + ("lastName", self.last_name), + ("dateOfBirth", self.date_of_birth), + ("memberId", self.member_id), + ): + if value: + subscriber[key] = value + encounter: dict[str, Any] = {"serviceTypeCodes": list(self.service_type_codes)} + encounter["dateOfService"] = self.date_of_service + return { + "tradingPartnerServiceId": self.payer_id, + "provider": provider, + "subscriber": subscriber, + "encounter": encounter, + } + + def safe_summary(self) -> dict[str, Any]: + """Non-PHI operational summary suitable for diagnostics.""" + return { + "operation_id": self.operation_id, + "payer_id": self.payer_id, + "service_type_codes": list(self.service_type_codes), + "date_of_service_present": self.date_of_service is not None, + } + + +class QualifiedBenefit(BaseModel): + code: str + value: str + value_kind: Literal["amount", "percent"] + service_type_codes: list[str] + network_code: Optional[str] = None + coverage_level_code: Optional[str] = None + time_qualifier_code: Optional[str] = None + procedure_code: Optional[str] = None + benefit_dates: dict[str, str] = Field(default_factory=dict) + + +class EligibilityResult(BaseModel): + model_config = ConfigDict(protected_namespaces=()) + + status: EligibilityStatus + payer_id: str + operation_id: str + application_mode: Optional[ApplicationMode] = None + payer_name: Optional[str] = None + plan_name: Optional[str] = None + plan_begin: Optional[str] = None + plan_end: Optional[str] = None + coverage_by_service: dict[str, EligibilityStatus] = Field(default_factory=dict) + benefits: list[QualifiedBenefit] = Field(default_factory=list) + copay: Optional[str] = None + coinsurance_percent: Optional[str] = None + deductible_total: Optional[str] = None + deductible_remaining: Optional[str] = None + out_of_pocket_total: Optional[str] = None + out_of_pocket_remaining: Optional[str] = None + service_type_codes: list[str] = Field(default_factory=list) + aaa_codes: list[str] = Field(default_factory=list) + ambiguities: list[str] = Field(default_factory=list) + error_category: Optional[ErrorCategory] = None + retry_disposition: RetryDisposition = RetryDisposition.NO_RETRY_QUEUE + reason: str = "" + raw_271: Optional[dict[str, Any]] = Field(default=None, exclude=True, repr=False) + raw_271_sha256: Optional[str] = None + raw_271_bytes: Optional[bytes] = Field(default=None, exclude=True, repr=False) + response_subject_sha256: Optional[str] = None + http_status: Optional[int] = None + checked_at: str = "" + source: str = "stedi" + attempt_count: int = 1 + request_sha256: str + + @field_validator("request_sha256", "raw_271_sha256", "response_subject_sha256") + @classmethod + def _digest(cls, value: Optional[str]) -> Optional[str]: + if value is not None and not re.fullmatch(r"[0-9a-f]{64}", value): + raise ValueError("eligibility digests must be lowercase SHA-256") + return value + + @property + def is_answer(self) -> bool: + return ( + self.status in (EligibilityStatus.ACTIVE, EligibilityStatus.INACTIVE) + and not self.ambiguities + and self.error_category is None + and isinstance(self.application_mode, ApplicationMode) + and self.response_subject_sha256 is not None + and self.http_status is not None + and 200 <= self.http_status < 300 + and self.raw_271_bytes is not None + and self.raw_271_sha256 is not None + and hashlib.sha256(self.raw_271_bytes).hexdigest() == self.raw_271_sha256 + ) + + @property + def retryable(self) -> bool: + return self.retry_disposition in ( + RetryDisposition.RETRY_THEN_PORTAL, + RetryDisposition.RETRY_THEN_QUEUE, + ) + + @property + def portal_fallback_allowed(self) -> bool: + return self.retry_disposition is RetryDisposition.RETRY_THEN_PORTAL + + +def _utcnow() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def eligibility_request_sha256(request: EligibilityRequest) -> str: + """Digest the exact JSON request body without persisting its PHI.""" + payload = json.dumps( + request.to_stedi_body(), sort_keys=True, separators=(",", ":") + ).encode("utf-8") + return hashlib.sha256(payload).hexdigest() + + +def _normalized_identity_text(value: object) -> str: + """Conservatively normalize identity text without dropping punctuation.""" + return " ".join(unicodedata.normalize("NFKC", str(value)).split()).casefold() + + +def _verify_response_subject( + body: Mapping[str, Any], request: EligibilityRequest +) -> tuple[Optional[str], Optional[str]]: + """Bind a subscriber-only request to the exact subject returned by the payer. + + The current request schema deliberately has no dependent field. A response + containing dependent results is therefore not silently interpreted as the + subscriber's answer. The member ID is mandatory; every additional identity + field supplied in the request must also be echoed and match after conservative + Unicode/case/whitespace normalization. + """ + dependents = body.get("dependents") + if dependents not in (None, []): + return None, "response contains dependent subjects for a subscriber request" + if dependents is not None and not isinstance(dependents, list): + return None, "response dependent subject collection is malformed" + + subscriber = body.get("subscriber") + if not isinstance(subscriber, dict): + return None, "response subscriber identity is missing or malformed" + + request_member = _normalized_identity_text(request.member_id) + response_member_raw = subscriber.get("memberId") + if response_member_raw is None: + return None, "response subscriber member ID is missing" + response_member = _normalized_identity_text(response_member_raw) + if not response_member or response_member != request_member: + return None, "response subscriber member ID does not match request" + + for request_field, response_field, label in ( + (request.first_name, "firstName", "first name"), + (request.last_name, "lastName", "last name"), + (request.date_of_birth, "dateOfBirth", "date of birth"), + ): + if request_field is None: + continue + response_value = subscriber.get(response_field) + if response_value is None: + return None, f"response subscriber {label} is missing" + expected = _normalized_identity_text(request_field) + observed = _normalized_identity_text(response_value) + if not observed or observed != expected: + return None, f"response subscriber {label} does not match request" + return "subscriber", None + + +def _verify_response_provider( + body: Mapping[str, Any], request: EligibilityRequest +) -> Optional[str]: + provider = body.get("provider") + if not isinstance(provider, dict): + return "response provider identity is missing or malformed" + response_npi = provider.get("npi") + if response_npi is None or str(response_npi) != request.provider_npi: + return "response provider NPI does not match request" + return None + + +def _benefit_entries(body: Mapping[str, Any]) -> list[dict[str, Any]]: + entries = body.get("benefitsInformation") + return ( + [e for e in entries if isinstance(e, dict)] if isinstance(entries, list) else [] + ) + + +def _collect_aaa_codes(body: Mapping[str, Any]) -> list[str]: + codes: list[str] = [] + errors = body.get("errors") + if isinstance(errors, list): + for error in errors: + if isinstance(error, dict) and error.get("code") is not None: + code = str(error["code"]) + if code not in codes: + codes.append(code) + return codes + + +def _date_value(value: str) -> Optional[date]: + try: + return datetime.strptime(value, "%Y%m%d").date() + except (TypeError, ValueError): + return None + + +def _date_range(value: str) -> tuple[Optional[date], Optional[date]]: + if "-" in value: + begin, end = value.split("-", 1) + return _date_value(begin), _date_value(end) + one = _date_value(value) + return one, one + + +def _coverage_interval( + dates: Mapping[str, Any], service_date: date +) -> tuple[Optional[bool], Optional[str]]: + starts: list[date] = [] + ends: list[date] = [] + recognized = False + for key, raw in dates.items(): + is_range = key in {"benefit", "plan", "eligibility", "policy"} + is_begin = key.lower().endswith(("begin", "effective")) + is_end = key.lower().endswith(("end", "expiration")) + if not (is_range or is_begin or is_end): + continue + recognized = True + if not isinstance(raw, str): + return None, "returned coverage interval is malformed" + if is_range: + begin, end = _date_range(raw) + if begin is None or end is None or begin > end: + return None, "returned coverage interval is malformed" + if begin: + starts.append(begin) + if end: + ends.append(end) + elif is_begin: + parsed = _date_value(raw) + if parsed is None: + return None, "returned coverage interval is malformed" + starts.append(parsed) + elif is_end: + parsed = _date_value(raw) + if parsed is None: + return None, "returned coverage interval is malformed" + ends.append(parsed) + if not recognized or (not starts and not ends): + return None, None + if starts and ends and max(starts) > min(ends): + return None, "returned coverage interval is contradictory" + return ( + (not starts or service_date >= max(starts)) + and (not ends or service_date <= min(ends)), + None, + ) + + +def _covers_service_date( + dates: Mapping[str, Any], service_date: date +) -> Optional[bool]: + covered, _problem = _coverage_interval(dates, service_date) + return covered + + +def _coverage_by_service( + body: Mapping[str, Any], request: EligibilityRequest +) -> tuple[dict[str, EligibilityStatus], list[str]]: + entries = _benefit_entries(body) + plan_status = body.get("planStatus") + if isinstance(plan_status, list): + entries = entries + [e for e in plan_status if isinstance(e, dict)] + service_date = ( + _date_value(request.date_of_service) if request.date_of_service else None + ) + output: dict[str, EligibilityStatus] = {} + problems: list[str] = [] + for service in request.service_type_codes: + matching = [ + e + for e in entries + if isinstance(e.get("serviceTypeCodes"), list) + and service in [str(v) for v in e["serviceTypeCodes"]] + ] + codes = {str(e.get("code", e.get("statusCode", ""))) for e in matching} + active = _CODE_ACTIVE in codes + inactive = bool(codes & _CODES_INACTIVE) + if active and inactive: + problems.append(f"service {service}: conflicting active/inactive coverage") + continue + if not active and not inactive: + problems.append(f"service {service}: no exact coverage signal") + continue + status = EligibilityStatus.ACTIVE if active else EligibilityStatus.INACTIVE + if status is EligibilityStatus.ACTIVE and service_date: + interval_results: list[bool] = [] + for entry in matching: + if str(entry.get("code", entry.get("statusCode", ""))) != _CODE_ACTIVE: + continue + info = entry.get("benefitsDateInformation") + if isinstance(info, dict): + covered, interval_problem = _coverage_interval(info, service_date) + if interval_problem: + problems.append(f"service {service}: {interval_problem}") + continue + if covered is not None: + interval_results.append(covered) + if not interval_results: + plan_dates = body.get("planDateInformation") + if isinstance(plan_dates, dict): + covered, interval_problem = _coverage_interval( + plan_dates, service_date + ) + if interval_problem: + problems.append(f"service {service}: {interval_problem}") + elif covered is not None: + interval_results.append(covered) + if not interval_results: + problems.append( + f"service {service}: no returned active coverage interval" + ) + continue + if not all(interval_results): + problems.append( + f"service {service}: service date outside returned benefit dates" + ) + continue + output[service] = status + return output, problems + + +def _qualified_benefits(body: Mapping[str, Any]) -> list[QualifiedBenefit]: + output: list[QualifiedBenefit] = [] + for entry in _benefit_entries(body): + code = str(entry.get("code", "")) + value_key = "benefitPercent" if code == "A" else "benefitAmount" + if code not in {"A", "B", "C", "G"} or entry.get(value_key) is None: + continue + procedure = entry.get("compositeMedicalProcedureIdentifier") + procedure_code = ( + procedure.get("procedureCode") if isinstance(procedure, dict) else None + ) + date_info = entry.get("benefitsDateInformation") + output.append( + QualifiedBenefit( + code=code, + value=str(entry[value_key]), + value_kind="percent" if value_key == "benefitPercent" else "amount", + service_type_codes=[str(v) for v in entry.get("serviceTypeCodes", [])], + network_code=( + str(entry["inPlanNetworkIndicatorCode"]) + if entry.get("inPlanNetworkIndicatorCode") is not None + else None + ), + coverage_level_code=( + str(entry["coverageLevelCode"]) + if entry.get("coverageLevelCode") is not None + else None + ), + time_qualifier_code=( + str(entry["timeQualifierCode"]) + if entry.get("timeQualifierCode") is not None + else None + ), + procedure_code=str(procedure_code) if procedure_code else None, + benefit_dates={ + str(k): str(v) + for k, v in date_info.items() + if isinstance(v, (str, int, float)) + } + if isinstance(date_info, dict) + else {}, + ) + ) + return output + + +def _selection_ambiguity( + benefits: list[QualifiedBenefit], request: EligibilityRequest +) -> Optional[str]: + """Refuse an answer when explicit benefit qualifiers match no evidence. + + Coverage status entries do not always carry network/coverage/time metadata. + When the caller declares one of those qualifiers, an unrelated active + service row must therefore not be enough to produce a consumable answer. + """ + selection = request.benefit_selection + requested = { + "network": selection.network_code, + "coverage level": selection.coverage_level_code, + "time period": selection.time_qualifier_code, + "procedure": selection.procedure_code, + } + explicit = {name: value for name, value in requested.items() if value is not None} + if not explicit: + return None + + service_date = _date_value(request.date_of_service) + service_candidates = [ + benefit + for benefit in benefits + if any( + service in benefit.service_type_codes + for service in request.service_type_codes + ) + ] + qualified = [ + benefit + for benefit in service_candidates + if ( + selection.network_code is None + or benefit.network_code in {selection.network_code, "W"} + ) + and ( + selection.coverage_level_code is None + or benefit.coverage_level_code == selection.coverage_level_code + ) + and ( + selection.time_qualifier_code is None + or benefit.time_qualifier_code == selection.time_qualifier_code + ) + and ( + selection.procedure_code is None + or benefit.procedure_code == selection.procedure_code + ) + and ( + service_date is None + or not benefit.benefit_dates + or _covers_service_date(benefit.benefit_dates, service_date) is True + ) + ] + if qualified: + return None + labels = ", ".join(sorted(explicit)) + return f"no patient-responsibility evidence matches requested {labels} qualifiers" + + +def _select_value( + benefits: list[QualifiedBenefit], + request: EligibilityRequest, + *, + code: str, + time_code: Optional[str] = None, +) -> tuple[Optional[str], Optional[str]]: + selection = request.benefit_selection + service_date = ( + _date_value(request.date_of_service) if request.date_of_service else None + ) + candidates = [ + b + for b in benefits + if b.code == code + and any( + service in b.service_type_codes for service in request.service_type_codes + ) + and (time_code is None or b.time_qualifier_code == time_code) + and ( + selection.time_qualifier_code is None + or time_code is not None + or b.time_qualifier_code == selection.time_qualifier_code + ) + and ( + selection.coverage_level_code is None + or b.coverage_level_code == selection.coverage_level_code + ) + and ( + selection.procedure_code is None + or b.procedure_code == selection.procedure_code + ) + and ( + selection.network_code is None + or b.network_code in {selection.network_code, "W"} + ) + and ( + service_date is None + or not b.benefit_dates + or _covers_service_date(b.benefit_dates, service_date) is True + ) + ] + if selection.network_code is not None: + exact = [b for b in candidates if b.network_code == selection.network_code] + if exact: + candidates = exact + values = sorted({b.value for b in candidates}) + if len(values) > 1: + suffix = f"/{time_code}" if time_code else "" + return None, f"benefit {code}{suffix}: conflicting qualified values" + return (values[0] if values else None), None + + +def _base_result( + request: EligibilityRequest, body_bytes: bytes, http_status: int +) -> dict[str, Any]: + return { + "payer_id": request.payer_id, + "operation_id": request.operation_id, + "service_type_codes": list(request.service_type_codes), + "raw_271_sha256": hashlib.sha256(body_bytes).hexdigest(), + "raw_271_bytes": body_bytes, + "http_status": http_status, + "checked_at": _utcnow(), + "request_sha256": eligibility_request_sha256(request), + } + + +def _error_result( + request: EligibilityRequest, + body_bytes: bytes, + *, + status: EligibilityStatus, + category: ErrorCategory, + disposition: RetryDisposition, + reason: str, + body: Optional[dict[str, Any]] = None, + aaa_codes: Optional[list[str]] = None, + http_status: int, +) -> EligibilityResult: + return EligibilityResult( + status=status, + error_category=category, + retry_disposition=disposition, + reason=reason, + raw_271=body, + aaa_codes=aaa_codes or [], + **_base_result(request, body_bytes, http_status), + ) + + +def parse_271( + request: EligibilityRequest, + body_bytes: bytes, + *, + http_status: int = 200, + expected_mode: Optional[ApplicationMode] = None, +) -> EligibilityResult: + """Normalize one response without ever choosing a first matching benefit.""" + # Transport status is authoritative even when an intermediary returns an + # HTML/text error body. In particular, a non-JSON 429 remains a bounded + # throttle retry rather than being mislabeled as an invalid response. + if http_status in {401, 403}: + return _error_result( + request, + body_bytes, + status=EligibilityStatus.REJECTED, + category=ErrorCategory.AUTH_CONFIGURATION, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: HTTP {http_status} authentication/configuration rejection", + http_status=http_status, + ) + if http_status == 429: + return _error_result( + request, + body_bytes, + status=EligibilityStatus.PAYER_UNAVAILABLE, + category=ErrorCategory.THROTTLED, + disposition=RetryDisposition.RETRY_THEN_PORTAL, + reason=f"payer {request.payer_id}: HTTP 429 request throttled before processing", + http_status=http_status, + ) + if http_status >= 500: + return _error_result( + request, + body_bytes, + status=EligibilityStatus.PAYER_UNAVAILABLE, + category=ErrorCategory.SERVER_TRANSIENT, + disposition=RetryDisposition.RETRY_THEN_PORTAL, + reason=f"payer {request.payer_id}: HTTP {http_status} clearinghouse/server failure", + http_status=http_status, + ) + if http_status < 200 or 300 <= http_status < 400: + return _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.RESPONSE_INVALID, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: HTTP {http_status} is not a successful response", + http_status=http_status, + ) + try: + decoded = json.loads(body_bytes.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + return _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.RESPONSE_INVALID, + disposition=( + RetryDisposition.RETRY_THEN_PORTAL + if http_status >= 500 + else RetryDisposition.NO_RETRY_QUEUE + ), + reason=f"payer {request.payer_id}: HTTP {http_status} non-JSON response", + http_status=http_status, + ) + if not isinstance(decoded, dict): + return _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.RESPONSE_INVALID, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: response JSON is not an object", + http_status=http_status, + ) + body = decoded + + aaa_codes = _collect_aaa_codes(body) + if http_status == 400 and "79" in aaa_codes: + category = ErrorCategory.INVALID_PAYER + elif http_status >= 400: + category = ErrorCategory.INVALID_REQUEST + else: + category = None + if category is not None: + return _error_result( + request, + body_bytes, + status=EligibilityStatus.REJECTED, + category=category, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: HTTP {http_status} {category.value}", + body=body, + aaa_codes=aaa_codes, + http_status=http_status, + ) + + response_payer = body.get("tradingPartnerServiceId") + if response_payer is None or str(response_payer) != request.payer_id: + return _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.INVALID_PAYER, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: response payer ID does not match request binding", + body=body, + aaa_codes=aaa_codes, + http_status=http_status, + ) + + meta = body.get("meta") + raw_mode = meta.get("applicationMode") if isinstance(meta, dict) else None + application_mode: Optional[ApplicationMode] = None + if raw_mode in {m.value for m in ApplicationMode}: + application_mode = ApplicationMode(raw_mode) + if application_mode is None or ( + expected_mode is not None and application_mode is not expected_mode + ): + return _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.AUTH_CONFIGURATION, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: response application mode does not match account boundary", + body=body, + aaa_codes=aaa_codes, + http_status=http_status, + ) + + if aaa_codes: + code_set = set(aaa_codes) + if code_set <= _AAA_TRANSIENT or code_set == {"42", "79"}: + category = ErrorCategory.PAYER_TRANSIENT + status = EligibilityStatus.PAYER_UNAVAILABLE + disposition = RetryDisposition.RETRY_THEN_PORTAL + elif code_set & _AAA_MEMBER: + category = ErrorCategory.MEMBER_IDENTITY + status = EligibilityStatus.NOT_FOUND + disposition = RetryDisposition.NO_RETRY_QUEUE + elif code_set & _AAA_PROVIDER: + category = ErrorCategory.PROVIDER_CONFIGURATION + status = EligibilityStatus.REJECTED + disposition = RetryDisposition.NO_RETRY_QUEUE + elif code_set & _AAA_INVALID_PAYER or code_set == {"79"}: + category = ErrorCategory.INVALID_PAYER + status = EligibilityStatus.REJECTED + disposition = RetryDisposition.NO_RETRY_QUEUE + else: + category = ErrorCategory.INVALID_REQUEST + status = EligibilityStatus.REJECTED + disposition = RetryDisposition.NO_RETRY_QUEUE + result = _error_result( + request, + body_bytes, + status=status, + category=category, + disposition=disposition, + reason=f"payer {request.payer_id}: AAA {','.join(aaa_codes)} classified {category.value}", + body=body, + aaa_codes=aaa_codes, + http_status=http_status, + ) + result.application_mode = application_mode + return result + + if body.get("error") is not None or body.get("errors"): + return _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.RESPONSE_INVALID, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: response includes an unclassified error", + body=body, + http_status=http_status, + ) + + subject_role, subject_problem = _verify_response_subject(body, request) + if subject_problem is not None: + return _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.MEMBER_IDENTITY, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: {subject_problem}", + body=body, + aaa_codes=aaa_codes, + http_status=http_status, + ) + assert subject_role == "subscriber" + provider_problem = _verify_response_provider(body, request) + if provider_problem is not None: + return _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.PROVIDER_CONFIGURATION, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: {provider_problem}", + body=body, + aaa_codes=aaa_codes, + http_status=http_status, + ) + response_subject_sha256 = hashlib.sha256( + b"openadapt.eligibility.subject.v1\0" + + eligibility_request_sha256(request).encode("ascii") + + b"\0" + + hashlib.sha256(body_bytes).hexdigest().encode("ascii") + + b"\0subscriber" + ).hexdigest() + + coverage, problems = _coverage_by_service(body, request) + statuses = set(coverage.values()) + if ( + problems + or len(coverage) != len(request.service_type_codes) + or len(statuses) != 1 + ): + result = _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.RESPONSE_AMBIGUOUS, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: requested-service coverage is incomplete or conflicting", + body=body, + aaa_codes=aaa_codes, + http_status=http_status, + ) + result.application_mode = application_mode + result.coverage_by_service = coverage + result.ambiguities = problems or [ + "requested services have mixed coverage states" + ] + return result + + status = next(iter(statuses)) + try: + benefits = _qualified_benefits(body) + except (TypeError, ValueError): + result = _error_result( + request, + body_bytes, + status=EligibilityStatus.INDETERMINATE, + category=ErrorCategory.RESPONSE_INVALID, + disposition=RetryDisposition.NO_RETRY_QUEUE, + reason=f"payer {request.payer_id}: qualified benefit structure is malformed", + body=body, + aaa_codes=aaa_codes, + http_status=http_status, + ) + result.application_mode = application_mode + return result + selections: dict[str, Optional[str]] = {} + selection_problem = _selection_ambiguity(benefits, request) + ambiguities: list[str] = [selection_problem] if selection_problem else [] + for field, code, time_code in ( + ("copay", "B", None), + ("coinsurance_percent", "A", None), + ("deductible_total", "C", "23"), + ("deductible_remaining", "C", "29"), + ("out_of_pocket_total", "G", "23"), + ("out_of_pocket_remaining", "G", "29"), + ): + value, problem = _select_value( + benefits, request, code=code, time_code=time_code + ) + selections[field] = value + if problem: + ambiguities.append(problem) + + payer = body.get("payer") + plan = body.get("planInformation") + dates = body.get("planDateInformation") + result = EligibilityResult( + status=status, + payer_id=request.payer_id, + operation_id=request.operation_id, + application_mode=application_mode, + payer_name=str(payer.get("name")) + if isinstance(payer, dict) and payer.get("name") + else None, + plan_name=( + str(plan.get("planDescription") or plan.get("groupDescription")) + if isinstance(plan, dict) + and (plan.get("planDescription") or plan.get("groupDescription")) + else None + ), + plan_begin=( + str(dates.get("planBegin") or dates.get("eligibilityBegin")) + if isinstance(dates, dict) + and (dates.get("planBegin") or dates.get("eligibilityBegin")) + else None + ), + plan_end=( + str(dates.get("planEnd") or dates.get("eligibilityEnd")) + if isinstance(dates, dict) + and (dates.get("planEnd") or dates.get("eligibilityEnd")) + else None + ), + coverage_by_service=coverage, + benefits=benefits, + ambiguities=ambiguities, + reason=f"payer {request.payer_id}: exact requested-service coverage parsed {status.value}", + raw_271=body, + raw_271_sha256=hashlib.sha256(body_bytes).hexdigest(), + raw_271_bytes=body_bytes, + response_subject_sha256=response_subject_sha256, + http_status=http_status, + checked_at=_utcnow(), + request_sha256=eligibility_request_sha256(request), + service_type_codes=list(request.service_type_codes), + copay=selections["copay"], + coinsurance_percent=selections["coinsurance_percent"], + deductible_total=selections["deductible_total"], + deductible_remaining=selections["deductible_remaining"], + out_of_pocket_total=selections["out_of_pocket_total"], + out_of_pocket_remaining=selections["out_of_pocket_remaining"], + ) + if ambiguities: + result.error_category = ErrorCategory.RESPONSE_AMBIGUOUS + result.reason = f"payer {request.payer_id}: coverage parsed but qualified benefit values conflict" + return result + + +class StediEligibilityClient: + """Bounded synchronous Stedi client with explicit account ownership.""" + + def __init__( + self, + *, + account: StediAccountBoundary, + auth: Optional[AuthRef] = None, + base_url: str = STEDI_ELIGIBILITY_URL, + timeout_s: float = 30.0, + max_attempts: int = 3, + max_response_bytes: int = MAX_RESPONSE_BYTES, + max_concurrency: int = 4, + transport: Any = None, + env: Optional[Mapping[str, str]] = None, + sleep: Callable[[float], None] = time.sleep, + ) -> None: + if base_url != STEDI_ELIGIBILITY_URL: + raise ValueError( + "eligibility endpoint is not the allowlisted Stedi endpoint" + ) + if timeout_s <= 0 or max_attempts < 1 or max_attempts > 5: + raise ValueError("timeout/max_attempts outside bounded policy") + if max_response_bytes < 1024 or max_concurrency < 1 or max_concurrency > 32: + raise ValueError("response/concurrency limit outside bounded policy") + self.account = account + self.base_url = base_url + self.timeout_s = timeout_s + self.max_attempts = max_attempts + self.max_response_bytes = max_response_bytes + self._transport = transport + self._sleep = sleep + self._semaphore = threading.BoundedSemaphore(max_concurrency) + self._headers = self._resolve_headers(auth, env) + + def _resolve_headers( + self, auth: Optional[AuthRef], env: Optional[Mapping[str, str]] + ) -> dict[str, str]: + if auth is not None: + headers = auth.resolve_headers(env) + else: + import os + + source = os.environ if env is None else env + key = source.get(self.account.credential_env, "") + if not key: + raise ValueError( + f"practice-held credential environment variable {self.account.credential_env!r} is empty" + ) + # Current Stedi docs prefer the bare key. The old ``Key `` prefix + # is accepted only for backwards compatibility and is not emitted. + if key.startswith("Key "): + key = key[4:] + headers = {"Authorization": key} + authorization = headers.get("Authorization", "") + if authorization.startswith("Key "): + authorization = authorization[4:] + if not authorization or "\r" in authorization or "\n" in authorization: + raise ValueError("practice-held Authorization credential is invalid") + headers["Authorization"] = authorization + headers.setdefault("Content-Type", "application/json") + return headers + + def _transport_failure( + self, request: EligibilityRequest, category: ErrorCategory, reason: str + ) -> EligibilityResult: + retryable = category in { + ErrorCategory.TRANSPORT_TRANSIENT, + ErrorCategory.THROTTLED, + ErrorCategory.SERVER_TRANSIENT, + ErrorCategory.PAYER_TRANSIENT, + } + return EligibilityResult( + status=( + EligibilityStatus.PAYER_UNAVAILABLE + if retryable + else EligibilityStatus.INDETERMINATE + ), + payer_id=request.payer_id, + operation_id=request.operation_id, + application_mode=self.account.application_mode, + service_type_codes=list(request.service_type_codes), + error_category=category, + retry_disposition=( + RetryDisposition.RETRY_THEN_PORTAL + if retryable + else RetryDisposition.NO_RETRY_QUEUE + ), + reason=reason, + checked_at=_utcnow(), + request_sha256=eligibility_request_sha256(request), + ) + + def _check_once(self, request: EligibilityRequest) -> EligibilityResult: + import httpx + + acquired = self._semaphore.acquire(timeout=self.timeout_s) + if not acquired: + return self._transport_failure( + request, + ErrorCategory.THROTTLED, + f"payer {request.payer_id}: local eligibility concurrency bound reached", + ) + try: + with httpx.Client( + transport=self._transport, timeout=self.timeout_s + ) as session: + try: + with session.stream( + "POST", + self.base_url, + headers=self._headers, + json=request.to_stedi_body(), + ) as response: + content_length = response.headers.get("content-length") + try: + declared_length = ( + int(content_length) if content_length else None + ) + except ValueError: + return self._transport_failure( + request, + ErrorCategory.RESPONSE_INVALID, + f"payer {request.payer_id}: invalid response content-length", + ) + if ( + declared_length is not None + and declared_length > self.max_response_bytes + ): + return self._transport_failure( + request, + ErrorCategory.RESPONSE_INVALID, + f"payer {request.payer_id}: response exceeds configured byte limit", + ) + payload = bytearray() + for chunk in response.iter_bytes(): + payload.extend(chunk) + if len(payload) > self.max_response_bytes: + return self._transport_failure( + request, + ErrorCategory.RESPONSE_INVALID, + f"payer {request.payer_id}: response exceeds configured byte limit", + ) + return parse_271( + request, + bytes(payload), + http_status=response.status_code, + expected_mode=self.account.application_mode, + ) + except (httpx.TimeoutException, httpx.NetworkError) as exc: + return self._transport_failure( + request, + ErrorCategory.TRANSPORT_TRANSIENT, + f"payer {request.payer_id}: transient transport {type(exc).__name__}", + ) + except httpx.HTTPError as exc: + return self._transport_failure( + request, + ErrorCategory.RESPONSE_INVALID, + f"payer {request.payer_id}: HTTP client failure {type(exc).__name__}", + ) + finally: + self._semaphore.release() + + def check(self, request: EligibilityRequest) -> EligibilityResult: + """Run a bounded retry loop for this idempotent 270 read. + + Stedi's eligibility endpoint does not advertise an idempotency-key + header. Retrying is nevertheless side-effect safe because 270 is a + read; ``operation_id`` binds all attempts to one local logical check. + """ + result: Optional[EligibilityResult] = None + for attempt in range(1, self.max_attempts + 1): + result = self._check_once(request) + result.attempt_count = attempt + if not result.retryable or attempt == self.max_attempts: + return result + self._sleep(min(0.25 * (2 ** (attempt - 1)), 2.0)) + assert result is not None + return result diff --git a/openadapt_flow/eligibility/payer_routes.yaml b/openadapt_flow/eligibility/payer_routes.yaml new file mode 100644 index 00000000..14f85691 --- /dev/null +++ b/openadapt_flow/eligibility/payer_routes.yaml @@ -0,0 +1,24 @@ +# Synthetic TEST-mode example only. Production route maps are practice-scoped +# deployment data. Each API route binds the exact request payer ID, stable Stedi +# payer ID, reviewed payer-record digest, supported service types, account, and +# application mode. Unknown or incomplete routes enter the attended queue. +version: 2 +default_route: queue +payers: + cigna_dental_mock: + display_name: Cigna Dental (Stedi test catalog) + route: api + aliases: [cigna dental test] + request_payer_id: "62308" + stedi_id: "HGJLR" + application_mode: test + practice_account_id: "openadapt-sandbox" + supported_service_type_codes: ["35"] + # SHA-256 of this canonical reviewed TEST record (without a trailing newline): + # {"coverageTypes":["dental"],"eligibilityCheck":"SUPPORTED","primaryPayerId":"62308","stediId":"HGJLR"} + # A live enrollment records the production payer record's own digest. + payer_record_sha256: "64ebef34642c651416428e365645a3fa5938cdc4a633cdab0af999fcbbe8f095" + portal_banned: false + portal_fallback_reviewed: false + verified_on: "2026-07-21" + notes: Stedi's published dental TEST request; no real member data. diff --git a/openadapt_flow/eligibility/waterfall.py b/openadapt_flow/eligibility/waterfall.py new file mode 100644 index 00000000..23e6f88b --- /dev/null +++ b/openadapt_flow/eligibility/waterfall.py @@ -0,0 +1,321 @@ +"""Exact, practice-scoped eligibility route resolution. + +The shipped YAML is a synthetic TEST-mode example. Production payer maps are +deployment data/recipes: the public engine supplies the schema and fail-closed +resolver, while each practice supplies reviewed bindings from Stedi's Payer +Network and its own portal policy. +""" + +from __future__ import annotations + +import re +from datetime import date +from enum import Enum +from pathlib import Path +from typing import Optional, Union + +from pydantic import BaseModel, Field, field_validator, model_validator + +from openadapt_flow.eligibility.client import ( + ApplicationMode, + EligibilityRequest, + EligibilityResult, + StediEligibilityClient, +) + +DEFAULT_REGISTRY_PATH = Path(__file__).parent / "payer_routes.yaml" +_SHA256_RE = re.compile(r"^[0-9a-f]{64}$") + + +class EligibilityRoute(str, Enum): + API = "api" + PORTAL = "portal" + QUEUE = "queue" + EXCLUDED = "excluded" + + +class PayerCapability(BaseModel): + """Reviewed route binding for one exact payer in one practice account.""" + + key: str + display_name: str = "" + route: EligibilityRoute + aliases: list[str] = Field(default_factory=list) + request_payer_id: Optional[str] = None + stedi_id: Optional[str] = None + application_mode: Optional[ApplicationMode] = None + practice_account_id: Optional[str] = None + supported_service_type_codes: list[str] = Field(default_factory=list) + payer_record_sha256: Optional[str] = None + portal_banned: bool = False + portal_fallback_reviewed: bool = False + verified_on: Optional[str] = None + notes: str = "" + + @field_validator("verified_on") + @classmethod + def _verification_date(cls, value: Optional[str]) -> Optional[str]: + if value is not None: + date.fromisoformat(value) + return value + + @model_validator(mode="after") + def _api_binding_complete(self) -> "PayerCapability": + if self.route is EligibilityRoute.PORTAL and self.portal_banned: + raise ValueError("a portal-banned payer cannot use the portal route") + if self.portal_banned and self.portal_fallback_reviewed: + raise ValueError("a portal-banned payer cannot approve portal fallback") + if self.route is EligibilityRoute.API: + missing = [ + name + for name, value in ( + ("request_payer_id", self.request_payer_id), + ("stedi_id", self.stedi_id), + ("application_mode", self.application_mode), + ("practice_account_id", self.practice_account_id), + ("supported_service_type_codes", self.supported_service_type_codes), + ("payer_record_sha256", self.payer_record_sha256), + ("verified_on", self.verified_on), + ) + if not value + ] + if missing: + raise ValueError( + f"API route lacks exact reviewed binding fields: {missing}" + ) + if not re.fullmatch(r"[A-Z0-9]{5}", self.stedi_id or ""): + raise ValueError("stedi_id must be the stable five-character Stedi ID") + if not _SHA256_RE.fullmatch(self.payer_record_sha256 or ""): + raise ValueError( + "payer_record_sha256 must bind the reviewed directory record" + ) + return self + + +class PayerRegistry(BaseModel): + version: int = 2 + default_route: EligibilityRoute = EligibilityRoute.QUEUE + payers: dict[str, PayerCapability] = Field(default_factory=dict) + + @model_validator(mode="after") + def _unique_names(self) -> "PayerRegistry": + if self.default_route is not EligibilityRoute.QUEUE: + raise ValueError("unmatched payers must default to the attended queue") + seen: dict[str, str] = {} + for key, cap in self.payers.items(): + for name in ( + key, + cap.key, + cap.display_name, + cap.request_payer_id, + cap.stedi_id, + *cap.aliases, + ): + if name is None: + continue + normalized = _normalize(name) + if not normalized: + continue + previous = seen.get(normalized) + if previous is not None and previous != key: + raise ValueError( + f"payer registry name/alias {name!r} maps to both {previous!r} and {key!r}" + ) + seen[normalized] = key + return self + + def find(self, payer: str) -> Optional[PayerCapability]: + wanted = _normalize(payer) + if not wanted: + return None + for cap in self.payers.values(): + if any( + _normalize(name) == wanted + for name in ( + cap.key, + cap.display_name, + cap.request_payer_id, + cap.stedi_id, + *cap.aliases, + ) + if name + ): + return cap + return None + + +class RouteDecision(BaseModel): + route: EligibilityRoute + payer: str + capability: Optional[PayerCapability] = None + reason: str = "" + + @property + def use_api(self) -> bool: + return self.route is EligibilityRoute.API + + +class WaterfallOutcome(BaseModel): + final_route: EligibilityRoute + decision: RouteDecision + result: Optional[EligibilityResult] = None + trail: list[str] = Field(default_factory=list) + + @property + def answered_by_api(self) -> bool: + return self.result is not None and self.result.is_answer + + +def _normalize(name: str) -> str: + return re.sub(r"[^a-z0-9]+", " ", name.lower()).strip() + + +def load_payer_routes(path: Union[str, Path, None] = None) -> PayerRegistry: + """Load and fully validate a route registry; malformed input fails loud.""" + import yaml + + registry_path = Path(path) if path is not None else DEFAULT_REGISTRY_PATH + raw = yaml.safe_load(registry_path.read_text(encoding="utf-8")) + if not isinstance(raw, dict) or not isinstance(raw.get("payers"), dict): + raise ValueError("payer route registry must contain a payer mapping") + payers: dict[str, PayerCapability] = {} + for key, entry in raw["payers"].items(): + if not isinstance(entry, dict): + raise ValueError(f"payer route entry {key!r} is not a mapping") + payers[str(key)] = PayerCapability(key=str(key), **entry) + return PayerRegistry( + version=int(raw.get("version", 2)), + default_route=EligibilityRoute(raw.get("default_route", "queue")), + payers=payers, + ) + + +def resolve_route( + payer: str, registry: Optional[PayerRegistry] = None +) -> RouteDecision: + reg = registry if registry is not None else load_payer_routes() + cap = reg.find(payer) + if cap is None: + return RouteDecision( + route=EligibilityRoute.QUEUE, + payer=payer, + reason=f"payer {payer!r} has no exact reviewed route binding; queue for resolution", + ) + return RouteDecision( + route=cap.route, + payer=payer, + capability=cap, + reason=f"payer {payer!r} resolved to reviewed {cap.route.value} route {cap.key!r}", + ) + + +def _queue( + decision: RouteDecision, + trail: list[str], + reason: str, + result: Optional[EligibilityResult] = None, +) -> WaterfallOutcome: + trail.append(reason) + return WaterfallOutcome( + final_route=EligibilityRoute.QUEUE, + decision=decision, + result=result, + trail=trail, + ) + + +def run_waterfall( + request: EligibilityRequest, + *, + payer: Optional[str] = None, + client: Optional[StediEligibilityClient] = None, + registry: Optional[PayerRegistry] = None, +) -> WaterfallOutcome: + """Run one exact route decision and bounded idempotent API attempt.""" + decision = resolve_route(payer or request.payer_id, registry) + trail = [decision.reason] + cap = decision.capability + if decision.route is EligibilityRoute.QUEUE: + return WaterfallOutcome( + final_route=decision.route, decision=decision, trail=trail + ) + if decision.route is EligibilityRoute.EXCLUDED: + return WaterfallOutcome( + final_route=decision.route, decision=decision, trail=trail + ) + if decision.route is EligibilityRoute.PORTAL: + return WaterfallOutcome( + final_route=decision.route, decision=decision, trail=trail + ) + assert cap is not None + + if request.payer_id != cap.request_payer_id: + return _queue( + decision, + trail, + "request payer ID does not exactly match the reviewed route binding", + ) + unsupported = set(request.service_type_codes) - set( + cap.supported_service_type_codes + ) + if unsupported: + return _queue( + decision, + trail, + "requested service type is not in the reviewed payer binding", + ) + if client is None: + return _queue(decision, trail, "API route has no practice-held client boundary") + if ( + client.account.practice_account_id != cap.practice_account_id + or client.account.application_mode is not cap.application_mode + ): + return _queue( + decision, + trail, + "client account/mode does not match the reviewed payer binding", + ) + + result = client.check(request) + trail.append(result.reason) + if result.is_answer: + return WaterfallOutcome( + final_route=EligibilityRoute.API, + decision=decision, + result=result, + trail=trail, + ) + if ( + result.portal_fallback_allowed + and cap.portal_fallback_reviewed + and not cap.portal_banned + ): + trail.append( + "bounded transient API attempts exhausted; use the reviewed portal fallback" + ) + return WaterfallOutcome( + final_route=EligibilityRoute.PORTAL, + decision=decision, + result=result, + trail=trail, + ) + if cap.portal_banned: + return _queue( + decision, + trail, + "portal fallback is barred; route the evidence-bearing halt to practice staff", + result, + ) + if result.portal_fallback_allowed and not cap.portal_fallback_reviewed: + return _queue( + decision, + trail, + "transient API attempts exhausted, but no portal fallback is explicitly reviewed", + result, + ) + return _queue( + decision, + trail, + "non-transient or ambiguous API outcome requires practice review", + result, + ) diff --git a/public-artifacts.json b/public-artifacts.json index 4f3ccb98..e874160c 100644 --- a/public-artifacts.json +++ b/public-artifacts.json @@ -1626,6 +1626,10 @@ "path": "openadapt_flow/console/static/index.html", "sha256": "388d3b60e1e8efe6542b530a4c8cdd3cdcd7ab925a392074d885a61e8d47ca90" }, + { + "path": "openadapt_flow/eligibility/payer_routes.yaml", + "sha256": "52c620badc980013fbf95bd910d2ebca25c9f0fbbaf1dfe86ea6771446343041" + }, { "path": "openadapt_flow/mockloan/static/app.js", "sha256": "5580accb8f4ecd94b48cc452ec84941da3814f0b1665bf8d239f2c241303765a" diff --git a/tests/test_eligibility_api.py b/tests/test_eligibility_api.py new file mode 100644 index 00000000..65b2b9ca --- /dev/null +++ b/tests/test_eligibility_api.py @@ -0,0 +1,645 @@ +"""Contract and fail-closed tests for the Stedi eligibility client.""" + +from __future__ import annotations + +import json + +import httpx +import pytest + +from openadapt_flow.eligibility.client import ( + STEDI_ELIGIBILITY_URL, + ApplicationMode, + BenefitSelection, + EligibilityRequest, + EligibilityStatus, + ErrorCategory, + RetryDisposition, + StediAccountBoundary, + StediEligibilityClient, + parse_271, +) + +ENV = {"STEDI_API_KEY": "test-key-123"} +ACCOUNT = StediAccountBoundary( + practice_account_id="openadapt-sandbox", + application_mode=ApplicationMode.TEST, +) + + +def dental_request(**overrides) -> EligibilityRequest: + base = dict( + operation_id="check-001", + payer_id="62308", + member_id="U3141592653", + first_name="Jaguar", + last_name="Dent", + date_of_birth="19960505", + provider_npi="1999999984", + provider_organization="One", + service_type_codes=["35"], + date_of_service="20260721", + benefit_selection=BenefitSelection(network_code="Y", coverage_level_code="IND"), + ) + base.update(overrides) + return EligibilityRequest(**base) + + +ACTIVE_271 = { + "meta": {"applicationMode": "test"}, + "tradingPartnerServiceId": "62308", + "subscriber": { + "memberId": "U3141592653", + "firstName": "Jaguar", + "lastName": "Dent", + "dateOfBirth": "19960505", + }, + "provider": {"npi": "1999999984"}, + "payer": {"name": "Cigna"}, + "planInformation": {"groupDescription": "DENTAL PPO"}, + "planDateInformation": {"planBegin": "20260101", "planEnd": "20261231"}, + "benefitsInformation": [ + {"code": "1", "serviceTypeCodes": ["35"]}, + { + "code": "B", + "benefitAmount": "20", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "Y", + }, + { + "code": "B", + "benefitAmount": "80", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "N", + }, + { + "code": "A", + "benefitPercent": "0.2", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "Y", + }, + { + "code": "C", + "benefitAmount": "1000", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "Y", + "timeQualifierCode": "23", + }, + { + "code": "C", + "benefitAmount": "500", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "Y", + "timeQualifierCode": "29", + }, + { + "code": "G", + "benefitAmount": "2500", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "Y", + "timeQualifierCode": "23", + }, + { + "code": "G", + "benefitAmount": "1200", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "Y", + "timeQualifierCode": "29", + }, + ], +} + + +def make_client(handler, **kwargs) -> StediEligibilityClient: + return StediEligibilityClient( + account=ACCOUNT, + transport=httpx.MockTransport(handler), + env=ENV, + sleep=lambda _: None, + **kwargs, + ) + + +def test_request_shape_and_phi_safe_repr(): + request = dental_request() + assert request.to_stedi_body()["encounter"] == { + "serviceTypeCodes": ["35"], + "dateOfService": "20260721", + } + assert request.to_stedi_body()["tradingPartnerServiceId"] == "62308" + assert "U3141592653" not in repr(request) + assert request.safe_summary() == { + "operation_id": "check-001", + "payer_id": "62308", + "service_type_codes": ["35"], + "date_of_service_present": True, + } + + +def test_date_of_service_is_required_for_a_deterministic_answer(): + with pytest.raises(ValueError, match="date_of_service"): + dental_request(date_of_service=None) + + +def test_client_uses_current_bare_authorization_header(): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["authorization"] = request.headers["Authorization"] + return httpx.Response(200, json=ACTIVE_271) + + assert make_client(handler).check(dental_request()).is_answer + assert seen["authorization"] == "test-key-123" + + +def test_legacy_key_prefix_is_removed_not_emitted(): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["authorization"] = request.headers["Authorization"] + return httpx.Response(200, json=ACTIVE_271) + + client = StediEligibilityClient( + account=ACCOUNT, + transport=httpx.MockTransport(handler), + env={"STEDI_API_KEY": "Key old-key"}, + ) + client.check(dental_request()) + assert seen["authorization"] == "old-key" + + +def test_missing_key_and_non_allowlisted_endpoint_fail_loud(): + with pytest.raises(ValueError, match="practice-held credential"): + StediEligibilityClient(account=ACCOUNT, env={}) + with pytest.raises(ValueError, match="allowlisted"): + StediEligibilityClient( + account=ACCOUNT, env=ENV, base_url="https://example.test/eligibility" + ) + with pytest.raises(ValueError, match="allowlisted"): + StediEligibilityClient( + account=ACCOUNT, + env=ENV, + base_url=f"{STEDI_ELIGIBILITY_URL}?alternate=true", + ) + + +def test_production_account_requires_practice_baa(): + with pytest.raises(ValueError, match="BAA"): + StediAccountBoundary( + practice_account_id="practice-1", + application_mode=ApplicationMode.PRODUCTION, + ) + + +def test_qualifier_aware_values_do_not_choose_out_of_network_or_remaining_as_total(): + payload = json.dumps(ACTIVE_271).encode() + result = parse_271(dental_request(), payload, expected_mode=ApplicationMode.TEST) + assert result.is_answer + assert result.status is EligibilityStatus.ACTIVE + assert result.copay == "20" + assert result.coinsurance_percent == "0.2" + assert result.deductible_total == "1000" + assert result.deductible_remaining == "500" + assert result.out_of_pocket_total == "2500" + assert result.out_of_pocket_remaining == "1200" + assert len(result.benefits) == 7 + assert result.raw_271_bytes == payload + + +@pytest.mark.parametrize( + ("selection", "mutations", "label"), + [ + ( + BenefitSelection(network_code="Y", coverage_level_code="IND"), + { + "inPlanNetworkIndicatorCode": "N", + "coverageLevelCode": "FAM", + }, + "coverage level, network", + ), + ( + BenefitSelection(time_qualifier_code="23"), + {"timeQualifierCode": "29"}, + "time period", + ), + ], +) +def test_explicit_qualifier_without_matching_evidence_never_answers( + selection, mutations, label +): + body = json.loads(json.dumps(ACTIVE_271)) + for entry in body["benefitsInformation"]: + if entry.get("code") in {"A", "B", "C", "G"}: + entry.update(mutations) + result = parse_271( + dental_request(benefit_selection=selection), + json.dumps(body).encode(), + expected_mode=ApplicationMode.TEST, + ) + assert result.status is EligibilityStatus.ACTIVE + assert not result.is_answer + assert result.error_category is ErrorCategory.RESPONSE_AMBIGUOUS + assert ( + f"no patient-responsibility evidence matches requested {label} qualifiers" + in result.ambiguities + ) + + +def test_raw_phi_response_is_excluded_from_repr_and_serialization(): + body = json.loads(json.dumps(ACTIVE_271)) + body["subscriber"] = {"memberId": "SECRET-MEMBER-123"} + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert not result.is_answer + assert result.error_category is ErrorCategory.MEMBER_IDENTITY + assert result.raw_271 is not None + assert "SECRET-MEMBER-123" not in repr(result) + assert "SECRET-MEMBER-123" not in result.model_dump_json() + + +def test_answer_requires_exact_raw_bytes_digest_and_application_mode(): + result = parse_271( + dental_request(), + json.dumps(ACTIVE_271).encode(), + expected_mode=ApplicationMode.TEST, + ) + assert result.is_answer + assert not result.model_copy(update={"raw_271_bytes": None}).is_answer + assert not result.model_copy(update={"raw_271_sha256": "0" * 64}).is_answer + assert not result.model_copy(update={"application_mode": None}).is_answer + assert not result.model_copy(update={"application_mode": "test"}).is_answer + + +@pytest.mark.parametrize( + ("field", "wrong"), + [ + ("memberId", "WRONG-MEMBER"), + ("firstName", "Wrong"), + ("lastName", "Wrong"), + ("dateOfBirth", "19800101"), + ], +) +def test_active_response_for_wrong_subject_never_answers(field, wrong): + body = json.loads(json.dumps(ACTIVE_271)) + body["subscriber"][field] = wrong + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert not result.is_answer + assert result.error_category is ErrorCategory.MEMBER_IDENTITY + assert "SECRET" not in result.reason + + +def test_subject_matching_is_conservative_but_case_and_whitespace_tolerant(): + body = json.loads(json.dumps(ACTIVE_271)) + body["subscriber"].update( + { + "memberId": " u3141592653 ", + "firstName": " JAGUAR ", + "lastName": "DENT", + } + ) + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert result.is_answer + assert result.response_subject_sha256 + + +def test_missing_or_dependent_response_subject_never_answers(): + missing = json.loads(json.dumps(ACTIVE_271)) + missing.pop("subscriber") + result = parse_271( + dental_request(), + json.dumps(missing).encode(), + expected_mode=ApplicationMode.TEST, + ) + assert not result.is_answer + assert result.error_category is ErrorCategory.MEMBER_IDENTITY + + dependent = json.loads(json.dumps(ACTIVE_271)) + dependent["dependents"] = [ + {"firstName": "Jaguar", "lastName": "Dent", "dateOfBirth": "19960505"} + ] + result = parse_271( + dental_request(), + json.dumps(dependent).encode(), + expected_mode=ApplicationMode.TEST, + ) + assert not result.is_answer + assert "dependent" in result.reason + + +@pytest.mark.parametrize("provider", [None, {"npi": "1000000000"}]) +def test_missing_or_wrong_response_provider_never_answers(provider): + body = json.loads(json.dumps(ACTIVE_271)) + if provider is None: + body.pop("provider") + else: + body["provider"] = provider + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert not result.is_answer + assert result.error_category is ErrorCategory.PROVIDER_CONFIGURATION + + +def test_response_application_mode_is_required_even_without_expected_mode(): + body = json.loads(json.dumps(ACTIVE_271)) + body.pop("meta") + result = parse_271(dental_request(), json.dumps(body).encode()) + assert not result.is_answer + assert result.error_category is ErrorCategory.AUTH_CONFIGURATION + + +@pytest.mark.parametrize( + ("status", "category", "retryable"), + [ + (401, ErrorCategory.AUTH_CONFIGURATION, False), + (429, ErrorCategory.THROTTLED, True), + (503, ErrorCategory.SERVER_TRANSIENT, True), + ], +) +def test_http_status_classification_does_not_depend_on_json_body( + status, category, retryable +): + result = parse_271(dental_request(), b"upstream text", http_status=status) + assert result.error_category is category + assert result.retryable is retryable + + +def test_conflicting_qualified_benefits_are_not_an_answer(): + body = json.loads(json.dumps(ACTIVE_271)) + body["benefitsInformation"].append( + { + "code": "B", + "benefitAmount": "25", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "Y", + } + ) + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert result.status is EligibilityStatus.ACTIVE + assert result.copay is None + assert not result.is_answer + assert result.error_category is ErrorCategory.RESPONSE_AMBIGUOUS + assert result.ambiguities == ["benefit B: conflicting qualified values"] + + +def test_wrong_service_and_conflicting_coverage_fail_closed(): + wrong = json.loads(json.dumps(ACTIVE_271)) + wrong["benefitsInformation"][0]["serviceTypeCodes"] = ["30"] + result = parse_271( + dental_request(), json.dumps(wrong).encode(), expected_mode=ApplicationMode.TEST + ) + assert result.status is EligibilityStatus.INDETERMINATE + assert not result.is_answer + + conflict = json.loads(json.dumps(ACTIVE_271)) + conflict["benefitsInformation"].append({"code": "6", "serviceTypeCodes": ["35"]}) + result = parse_271( + dental_request(), + json.dumps(conflict).encode(), + expected_mode=ApplicationMode.TEST, + ) + assert result.status is EligibilityStatus.INDETERMINATE + assert "conflicting active/inactive" in result.ambiguities[0] + + +def test_service_date_outside_plan_and_benefit_window_fails_closed(): + body = json.loads(json.dumps(ACTIVE_271)) + body["planDateInformation"]["planEnd"] = "20260630" + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert result.status is EligibilityStatus.INDETERMINATE + assert "service date" in result.ambiguities[0] + + +@pytest.mark.parametrize( + "plan_dates", + [ + {"planBegin": "20260722", "planEnd": "20261231"}, + {"planBegin": "20260101", "planEnd": "20260720"}, + {}, + ], +) +def test_active_answer_requires_explicit_interval_containing_service_date(plan_dates): + body = json.loads(json.dumps(ACTIVE_271)) + body["planDateInformation"] = plan_dates + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert not result.is_answer + assert result.error_category is ErrorCategory.RESPONSE_AMBIGUOUS + assert ( + "coverage interval" in result.ambiguities[0] + or "service date" in result.ambiguities[0] + ) + + +@pytest.mark.parametrize( + "plan_dates", + [ + {"planBegin": "20260101"}, + {"planEnd": "20261231"}, + ], +) +def test_active_answer_accepts_explicit_open_interval_containing_service_date( + plan_dates, +): + body = json.loads(json.dumps(ACTIVE_271)) + body["planDateInformation"] = plan_dates + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert result.is_answer + + +def test_redirect_with_valid_looking_json_is_never_an_answer(): + result = parse_271( + dental_request(), + json.dumps(ACTIVE_271).encode(), + http_status=302, + expected_mode=ApplicationMode.TEST, + ) + assert not result.is_answer + assert result.error_category is ErrorCategory.RESPONSE_INVALID + + +@pytest.mark.parametrize("status", [401, 403]) +def test_auth_errors_do_not_retry_or_fallback(status): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response(status, json={"error": "unauthorized"}) + + result = make_client(handler).check(dental_request()) + assert calls == 1 + assert result.error_category is ErrorCategory.AUTH_CONFIGURATION + assert result.retry_disposition is RetryDisposition.NO_RETRY_QUEUE + + +def test_invalid_response_is_queued_without_retry_or_portal_fallback(): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response(200, content=b"not-json") + + result = make_client(handler).check(dental_request()) + assert calls == 1 + assert result.error_category is ErrorCategory.RESPONSE_INVALID + assert result.status is EligibilityStatus.INDETERMINATE + assert result.retry_disposition is RetryDisposition.NO_RETRY_QUEUE + + +def test_http_400_aaa79_is_invalid_payer_and_not_retried(): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response(400, json={"errors": [{"code": "79"}]}) + + result = make_client(handler).check(dental_request()) + assert calls == 1 + assert result.error_category is ErrorCategory.INVALID_PAYER + assert not result.retryable + + +def test_aaa_member_identity_is_queued_not_automatically_retried(): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + body = { + "meta": {"applicationMode": "test"}, + "tradingPartnerServiceId": "62308", + "errors": [{"code": "75", "description": "contains no logged data"}], + } + return httpx.Response(200, json=body) + + result = make_client(handler).check(dental_request()) + assert calls == 1 + assert result.status is EligibilityStatus.NOT_FOUND + assert result.error_category is ErrorCategory.MEMBER_IDENTITY + assert "description" not in result.reason + + +@pytest.mark.parametrize( + "first_status,first_body", + [ + (429, {"code": "TOO_MANY_REQUESTS"}), + (503, {"error": "service_unavailable"}), + ( + 200, + { + "meta": {"applicationMode": "test"}, + "tradingPartnerServiceId": "62308", + "errors": [{"code": "42"}], + }, + ), + ], +) +def test_only_explicit_transients_retry(first_status, first_body): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + if calls == 1: + return httpx.Response(first_status, json=first_body) + return httpx.Response(200, json=ACTIVE_271) + + result = make_client(handler).check(dental_request()) + assert calls == 2 + assert result.is_answer + assert result.attempt_count == 2 + + +def test_transport_failure_is_bounded_and_secret_free(): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + raise httpx.ConnectError("member U3141592653 key test-key-123", request=request) + + result = make_client(handler, max_attempts=2).check(dental_request()) + assert calls == 2 + assert result.error_category is ErrorCategory.TRANSPORT_TRANSIENT + assert "U3141592653" not in result.reason + assert "test-key-123" not in result.reason + result.model_dump_json() + + +def test_response_size_is_bounded(): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=b"x" * 2049) + + result = make_client(handler, max_response_bytes=1024, max_attempts=1).check( + dental_request() + ) + assert result.error_category is ErrorCategory.RESPONSE_INVALID + assert not result.is_answer + + +def test_partial_result_with_unclassified_error_is_not_accepted(): + body = {**ACTIVE_271, "error": "partial_failure"} + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert not result.is_answer + assert result.error_category is ErrorCategory.RESPONSE_INVALID + + +def test_response_payer_and_mode_must_match_request_boundary(): + wrong_payer = {**ACTIVE_271, "tradingPartnerServiceId": "99999"} + result = parse_271( + dental_request(), + json.dumps(wrong_payer).encode(), + expected_mode=ApplicationMode.TEST, + ) + assert result.error_category is ErrorCategory.INVALID_PAYER + + wrong_mode = {**ACTIVE_271, "meta": {"applicationMode": "production"}} + result = parse_271( + dental_request(), + json.dumps(wrong_mode).encode(), + expected_mode=ApplicationMode.TEST, + ) + assert result.error_category is ErrorCategory.AUTH_CONFIGURATION + + +def test_non_json_and_empty_json_never_guess_active(): + for payload in (b"bad", b"{}"): + result = parse_271(dental_request(), payload) + assert not result.is_answer + assert result.status is EligibilityStatus.INDETERMINATE + + +def test_malformed_qualified_benefit_fails_closed_without_raising(): + body = json.loads(json.dumps(ACTIVE_271)) + body["benefitsInformation"].append( + {"code": "B", "benefitAmount": "20", "serviceTypeCodes": None} + ) + result = parse_271( + dental_request(), json.dumps(body).encode(), expected_mode=ApplicationMode.TEST + ) + assert not result.is_answer + assert result.error_category is ErrorCategory.RESPONSE_INVALID diff --git a/tests/test_eligibility_artifact.py b/tests/test_eligibility_artifact.py new file mode 100644 index 00000000..d58c8792 --- /dev/null +++ b/tests/test_eligibility_artifact.py @@ -0,0 +1,617 @@ +"""Atomic, idempotent, PHI-bound eligibility artifact tests.""" + +from __future__ import annotations + +import base64 +import csv +import json +import os +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +from openadapt_flow.eligibility.artifact import ( + BOUNDARY_FILE, + RESULTS_CSV, + TRANSACTIONS_DIR, + ArtifactEncryption, + PracticeArtifactPolicy, + all_confirmed, + purge_expired_eligibility_artifacts, + write_and_verify, + write_eligibility_artifacts, +) +from openadapt_flow.eligibility.client import ( + ApplicationMode, + BenefitSelection, + EligibilityRequest, + EligibilityStatus, + parse_271, +) +from openadapt_flow.runtime.effects.document_hash import DocumentHashVerifier +from openadapt_flow.runtime.effects.effect import Verdict + +ACTIVE = { + "meta": {"applicationMode": "test"}, + "tradingPartnerServiceId": "62308", + "subscriber": {"memberId": "U3141592653"}, + "provider": {"npi": "1999999984"}, + "payer": {"name": "Cigna"}, + "planInformation": {"groupDescription": "DENTAL PPO"}, + "planDateInformation": {"planBegin": "20260101", "planEnd": "20261231"}, + "benefitsInformation": [ + {"code": "1", "serviceTypeCodes": ["35"]}, + { + "code": "C", + "benefitAmount": "50", + "serviceTypeCodes": ["35"], + "coverageLevelCode": "IND", + "inPlanNetworkIndicatorCode": "Y", + "timeQualifierCode": "23", + }, + ], +} + + +def request(operation_id="artifact-check-1") -> EligibilityRequest: + return EligibilityRequest( + operation_id=operation_id, + payer_id="62308", + provider_npi="1999999984", + provider_organization="One", + member_id="U3141592653", + service_type_codes=["35"], + date_of_service="20260721", + benefit_selection=BenefitSelection(network_code="Y", coverage_level_code="IND"), + ) + + +def active_result(operation_id="artifact-check-1"): + return parse_271( + request(operation_id), + json.dumps(ACTIVE).encode(), + expected_mode=ApplicationMode.TEST, + ) + + +def volume_policy( + boundary="practice-1", mode=ApplicationMode.TEST +) -> PracticeArtifactPolicy: + return PracticeArtifactPolicy( + boundary_id=boundary, + application_mode=mode, + encryption=ArtifactEncryption.PLATFORM_VOLUME, + volume_encryption_attested=True, + retention_days=30, + ) + + +def encrypted_policy() -> PracticeArtifactPolicy: + return PracticeArtifactPolicy( + boundary_id="practice-encrypted", + application_mode=ApplicationMode.TEST, + encryption=ArtifactEncryption.APPLICATION_AES256_GCM, + encryption_key_env="ELIGIBILITY_ARTIFACT_KEY", + retention_days=30, + ) + + +def test_raw_and_normalized_records_promote_together_and_verify(tmp_path): + artifact, verdicts = write_and_verify( + active_result(), + tmp_path, + request=request(), + policy=volume_policy(), + ) + assert artifact.created + assert len(verdicts) == 4 + assert all_confirmed(verdicts), [v.reason for v in verdicts] + tx = Path(artifact.transaction_dir) + assert tx.is_dir() + assert {p.name for p in tx.iterdir()} == { + Path(artifact.raw_271_file).name, + Path(artifact.normalized_file).name, + "manifest.json", + } + manifest = json.loads((tx / "manifest.json").read_text()) + assert manifest["raw_plain_sha256"] == artifact.raw_271_sha256 + assert manifest["normalized_plain_sha256"] == artifact.normalized_sha256 + assert ( + manifest["response_subject_sha256"] == active_result().response_subject_sha256 + ) + assert manifest["egress"] == "none" + assert manifest["application_mode"] == "test" + assert manifest["http_status"] == 200 + assert manifest["committed_at"] == artifact.committed_at + assert manifest["retention_expires_at"].endswith("Z") + boundary = json.loads((tmp_path / BOUNDARY_FILE).read_text()) + assert boundary["application_mode"] == "test" + + +def test_csv_is_derived_and_formula_neutralized(tmp_path): + bound_request = request().model_copy(update={"member_id": '=HYPERLINK("bad")'}) + response = json.loads(json.dumps(ACTIVE)) + response["subscriber"]["memberId"] = bound_request.member_id + response["payer"]["name"] = "+malicious" + result = parse_271( + bound_request, + json.dumps(response).encode(), + expected_mode=ApplicationMode.TEST, + ) + artifact, verdicts = write_and_verify( + result, + tmp_path, + request=bound_request, + policy=volume_policy(), + ) + assert all_confirmed(verdicts) + with Path(artifact.results_csv).open(newline="") as handle: + rows = list(csv.DictReader(handle)) + assert len(rows) == 1 + assert rows[0]["member_id"].startswith("'=") + assert rows[0]["payer"].startswith("'+") + assert rows[0]["deductible_total"] == "50" + assert rows[0]["status"] == EligibilityStatus.ACTIVE.value + assert rows[0]["application_mode"] == "test" + assert rows[0]["http_status"] == "200" + assert rows[0]["committed_at"] + + +def test_same_operation_and_content_is_idempotent(tmp_path): + first, first_verdicts = write_and_verify( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + second, second_verdicts = write_and_verify( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + assert first.created and not second.created + assert first.transaction_dir == second.transaction_dir + assert all_confirmed(first_verdicts) and all_confirmed(second_verdicts) + with (tmp_path / RESULTS_CSV).open(newline="") as handle: + assert len(list(csv.DictReader(handle))) == 1 + + +def test_operation_id_reuse_with_changed_content_refuses(tmp_path): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + changed = {**ACTIVE, "controlNumber": "changed"} + changed_result = parse_271( + request(), json.dumps(changed).encode(), expected_mode=ApplicationMode.TEST + ) + with pytest.raises(FileExistsError, match="different content"): + write_eligibility_artifacts( + changed_result, tmp_path, request=request(), policy=volume_policy() + ) + + +def test_tampered_storage_is_refuted(tmp_path): + artifact, verdicts = write_and_verify( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + assert all_confirmed(verdicts) + raw = Path(artifact.raw_271_file) + raw.write_bytes(raw.read_bytes() + b"tampered") + verifier = DocumentHashVerifier(tmp_path, glob="transactions/tx_*/*") + state = verifier.capture_pre_state() + digest_effect = artifact.effects[1] + verdict = verifier.verify(digest_effect, state) + assert verdict.verdict is Verdict.REFUTED + + +def test_application_encryption_leaks_no_member_or_raw_payload(tmp_path): + key = base64.urlsafe_b64encode(b"k" * 32).decode() + artifact, verdicts = write_and_verify( + active_result(), + tmp_path, + request=request(), + policy=encrypted_policy(), + env={"ELIGIBILITY_ARTIFACT_KEY": key}, + ) + assert all_confirmed(verdicts) + assert artifact.results_csv.endswith(".enc") + for path in tmp_path.rglob("*"): + if path.is_file(): + payload = path.read_bytes() + assert b"U3141592653" not in payload + assert b"benefitsInformation" not in payload + + +def test_application_encryption_requires_real_32_byte_key(tmp_path): + with pytest.raises(ValueError, match="decode to 32 bytes"): + write_eligibility_artifacts( + active_result(), + tmp_path, + request=request(), + policy=encrypted_policy(), + env={ + "ELIGIBILITY_ARTIFACT_KEY": base64.urlsafe_b64encode(b"short").decode() + }, + ) + + +def test_boundary_policy_mismatch_fails_loud(tmp_path): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + with pytest.raises(ValueError, match="different PHI policy"): + write_eligibility_artifacts( + active_result("other-operation"), + tmp_path, + request=request("other-operation"), + policy=volume_policy("practice-2"), + ) + assert (tmp_path / BOUNDARY_FILE).exists() + + +def test_symlinked_root_and_index_are_refused(tmp_path): + outside = tmp_path / "outside" + outside.mkdir() + linked = tmp_path / "linked" + try: + linked.symlink_to(outside, target_is_directory=True) + except OSError: + pytest.skip("symlinks unavailable") + with pytest.raises(ValueError, match="not a link"): + write_eligibility_artifacts( + active_result(), linked, request=request(), policy=volume_policy() + ) + + root = tmp_path / "root" + write_eligibility_artifacts( + active_result(), root, request=request(), policy=volume_policy() + ) + (root / RESULTS_CSV).unlink() + (root / RESULTS_CSV).symlink_to(outside / "stolen.csv") + with pytest.raises(ValueError, match="symlinked"): + write_eligibility_artifacts( + active_result("next-operation"), + root, + request=request("next-operation"), + policy=volume_policy(), + ) + + +def test_existing_broad_directory_is_refused_without_chmod(tmp_path): + root = tmp_path / "broad" + root.mkdir(mode=0o755) + root.chmod(0o755) + with pytest.raises(PermissionError, match="already be owner-only"): + write_eligibility_artifacts( + active_result(), root, request=request(), policy=volume_policy() + ) + assert root.stat().st_mode & 0o777 == 0o755 + + +def test_concurrent_writer_lock_fails_fast(tmp_path): + tmp_path.chmod(0o700) + (tmp_path / ".eligibility-write.lock").mkdir() + with pytest.raises(BlockingIOError, match="holds the lock"): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + + +def test_atomic_promotion_failure_leaves_no_consumable_transaction( + tmp_path, monkeypatch +): + real_rename = os.rename + + def fail_stage(source, destination): + if ".staging-" in str(source): + raise OSError("injected rename failure") + return real_rename(source, destination) + + monkeypatch.setattr(os, "rename", fail_stage) + with pytest.raises(OSError, match="injected"): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + transactions = tmp_path / "transactions" + assert transactions.exists() + assert list(transactions.iterdir()) == [] + assert not (tmp_path / RESULTS_CSV).exists() + + +@pytest.mark.parametrize("target", ["raw", "normalized", "manifest", "index"]) +def test_every_staged_contract_is_verified_before_promotion( + tmp_path, monkeypatch, target +): + import openadapt_flow.eligibility.artifact as artifact_module + + real_write = artifact_module._write_exclusive + + def corrupt_after_write(path, payload): + real_write(path, payload) + name = path.name + selected = ( + (target == "raw" and name.startswith("raw_271_")) + or (target == "normalized" and name.startswith("result_")) + or (target == "manifest" and name == "manifest.json") + or (target == "index" and name.startswith(f".{RESULTS_CSV}.")) + ) + if selected: + with path.open("ab") as handle: + handle.write(b"corrupt") + + monkeypatch.setattr(artifact_module, "_write_exclusive", corrupt_after_write) + with pytest.raises((ValueError, json.JSONDecodeError)): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + transactions = tmp_path / TRANSACTIONS_DIR + assert transactions.is_dir() + assert list(transactions.iterdir()) == [] + assert not (tmp_path / RESULTS_CSV).exists() + assert not list(tmp_path.glob(".*.tmp")) + + +def test_index_promotion_failure_rolls_back_promoted_transaction(tmp_path, monkeypatch): + real_replace = os.replace + + def fail_index(source, destination): + if Path(destination).name == RESULTS_CSV: + raise OSError("injected index promotion failure") + return real_replace(source, destination) + + monkeypatch.setattr(os, "replace", fail_index) + with pytest.raises(OSError, match="index promotion"): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + assert list((tmp_path / TRANSACTIONS_DIR).iterdir()) == [] + assert not (tmp_path / RESULTS_CSV).exists() + + +def test_independent_effect_failure_rolls_back_new_transaction(tmp_path, monkeypatch): + import openadapt_flow.eligibility.artifact as artifact_module + + monkeypatch.setattr(artifact_module, "all_confirmed", lambda _verdicts: False) + with pytest.raises(RuntimeError, match="did not confirm"): + write_and_verify( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + assert list((tmp_path / TRANSACTIONS_DIR).iterdir()) == [] + with (tmp_path / RESULTS_CSV).open(newline="") as handle: + assert list(csv.DictReader(handle)) == [] + + +def test_expired_transaction_is_purged_and_cannot_be_reused(tmp_path): + policy = PracticeArtifactPolicy( + boundary_id="practice-retention", + application_mode=ApplicationMode.TEST, + encryption=ArtifactEncryption.PLATFORM_VOLUME, + volume_encryption_attested=True, + retention_days=1, + ) + artifact = write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=policy + ) + manifest = json.loads( + (Path(artifact.transaction_dir) / "manifest.json").read_text() + ) + expires = datetime.strptime( + manifest["retention_expires_at"], "%Y-%m-%dT%H:%M:%SZ" + ).replace(tzinfo=timezone.utc) + removed = purge_expired_eligibility_artifacts( + tmp_path, policy=policy, now=expires + timedelta(seconds=1) + ) + assert removed == [request().operation_id] + assert not Path(artifact.transaction_dir).exists() + with (tmp_path / RESULTS_CSV).open(newline="") as handle: + assert list(csv.DictReader(handle)) == [] + + replacement = write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=policy + ) + assert replacement.created + + +def test_retention_index_failure_restores_expired_transaction(tmp_path, monkeypatch): + policy = PracticeArtifactPolicy( + boundary_id="practice-retention-rollback", + application_mode=ApplicationMode.TEST, + encryption=ArtifactEncryption.PLATFORM_VOLUME, + volume_encryption_attested=True, + retention_days=1, + ) + artifact = write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=policy + ) + manifest = json.loads( + (Path(artifact.transaction_dir) / "manifest.json").read_text() + ) + expires = datetime.strptime( + manifest["retention_expires_at"], "%Y-%m-%dT%H:%M:%SZ" + ).replace(tzinfo=timezone.utc) + real_replace = os.replace + + def fail_index(source, destination): + if Path(destination).name == RESULTS_CSV: + raise OSError("injected retention index failure") + return real_replace(source, destination) + + monkeypatch.setattr(os, "replace", fail_index) + with pytest.raises(OSError, match="retention index"): + purge_expired_eligibility_artifacts( + tmp_path, policy=policy, now=expires + timedelta(seconds=1) + ) + assert Path(artifact.transaction_dir).is_dir() + with (tmp_path / RESULTS_CSV).open(newline="") as handle: + assert len(list(csv.DictReader(handle))) == 1 + + +@pytest.mark.parametrize( + ("expiry", "error"), + [ + ("not-a-timestamp", "retention expiry"), + ("2099-01-01T00:00:00Z", "bound policy"), + ], +) +def test_invalid_retention_manifest_denies_purge_without_deleting( + tmp_path, expiry, error +): + policy = volume_policy("practice-retention-denial") + artifact = write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=policy + ) + manifest_path = Path(artifact.transaction_dir) / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["retention_expires_at"] = expiry + manifest_path.write_text(json.dumps(manifest)) + with pytest.raises(ValueError, match=error): + purge_expired_eligibility_artifacts( + tmp_path, + policy=policy, + now=datetime.now(timezone.utc) + timedelta(days=365), + ) + assert Path(artifact.transaction_dir).is_dir() + + +def test_transport_outcome_without_raw_response_is_not_promoted(tmp_path): + result = active_result().model_copy( + update={"raw_271_bytes": None, "raw_271_sha256": None} + ) + with pytest.raises(ValueError, match="raw response"): + write_eligibility_artifacts( + result, tmp_path, request=request(), policy=volume_policy() + ) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("copay", "999999"), + ("payer_name", "Fabricated Payer"), + ("response_subject_sha256", "0" * 64), + ], +) +def test_mutated_normalized_result_cannot_be_promoted(tmp_path, field, value): + result = active_result().model_copy(update={field: value}) + with pytest.raises(ValueError, match="does not match the exact raw response"): + write_eligibility_artifacts( + result, tmp_path, request=request(), policy=volume_policy() + ) + + +def test_non_answer_and_wrong_subject_binding_are_not_consumable(tmp_path): + ambiguous_body = json.loads(json.dumps(ACTIVE)) + ambiguous_body["benefitsInformation"].append( + {"code": "6", "serviceTypeCodes": ["35"]} + ) + ambiguous = parse_271( + request(), + json.dumps(ambiguous_body).encode(), + expected_mode=ApplicationMode.TEST, + ) + with pytest.raises(ValueError, match="unambiguous"): + write_eligibility_artifacts( + ambiguous, tmp_path / "ambiguous", request=request(), policy=volume_policy() + ) + + wrong_subject = request().model_copy(update={"member_id": "OTHER-MEMBER"}) + with pytest.raises(ValueError, match="not bound"): + write_eligibility_artifacts( + active_result(), + tmp_path / "wrong-subject", + request=wrong_subject, + policy=volume_policy(), + ) + + +def test_tampered_manifest_cannot_escape_transaction_directory(tmp_path): + artifact = write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + manifest_path = Path(artifact.transaction_dir) / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["normalized_file"] = "../../../outside.json" + manifest_path.write_text(json.dumps(manifest)) + with pytest.raises(ValueError, match="normalized_file"): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + + +def test_test_answer_cannot_enter_a_production_artifact_boundary(tmp_path): + with pytest.raises(ValueError, match="application mode"): + write_eligibility_artifacts( + active_result(), + tmp_path, + request=request(), + policy=volume_policy(mode=ApplicationMode.PRODUCTION), + ) + assert not (tmp_path / TRANSACTIONS_DIR).exists() + + +def test_artifact_boundary_cannot_mix_application_modes(tmp_path): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + production_response = json.loads(json.dumps(ACTIVE)) + production_response["meta"]["applicationMode"] = "production" + production_request = request("production-check") + production_result = parse_271( + production_request, + json.dumps(production_response).encode(), + expected_mode=ApplicationMode.PRODUCTION, + ) + assert production_result.is_answer + with pytest.raises(ValueError, match="different PHI policy"): + write_eligibility_artifacts( + production_result, + tmp_path, + request=production_request, + policy=volume_policy(mode=ApplicationMode.PRODUCTION), + ) + + +def test_caller_timestamp_is_not_persisted_as_audit_truth(tmp_path): + result = active_result().model_copy(update={"checked_at": "1900-01-01T00:00:00Z"}) + artifact = write_eligibility_artifacts( + result, tmp_path, request=request(), policy=volume_policy() + ) + normalized = json.loads(Path(artifact.normalized_file).read_text()) + assert "checked_at" not in normalized + assert normalized["committed_at"] == artifact.committed_at + assert normalized["committed_at"] != result.checked_at + assert normalized["application_mode"] == "test" + assert normalized["http_status"] == 200 + + +@pytest.mark.parametrize("target", ["transaction", "raw", "normalized", "manifest"]) +def test_committed_paths_must_remain_owner_only(tmp_path, target): + artifact = write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + tx = Path(artifact.transaction_dir) + paths = { + "transaction": tx, + "raw": Path(artifact.raw_271_file), + "normalized": Path(artifact.normalized_file), + "manifest": tx / "manifest.json", + } + paths[target].chmod(0o755 if target == "transaction" else 0o644) + with pytest.raises(PermissionError, match="owner-only"): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + + +def test_committed_file_symlink_substitution_is_refused(tmp_path): + artifact = write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) + raw = Path(artifact.raw_271_file) + outside = tmp_path.parent / f"{tmp_path.name}-outside.json" + outside.write_bytes(raw.read_bytes()) + raw.unlink() + try: + raw.symlink_to(outside) + except OSError: + pytest.skip("symlinks unavailable") + with pytest.raises((OSError, ValueError)): + write_eligibility_artifacts( + active_result(), tmp_path, request=request(), policy=volume_policy() + ) diff --git a/tests/test_eligibility_live_stedi.py b/tests/test_eligibility_live_stedi.py new file mode 100644 index 00000000..d948fda0 --- /dev/null +++ b/tests/test_eligibility_live_stedi.py @@ -0,0 +1,84 @@ +"""Three-trial live Stedi TEST-mode qualification. + +The request is Stedi's published Cigna dental mock (STC 35), so it uses no +real member data and Stedi documents mock checks as free. Without a test key, +all three trials skip with one reproducible setup instruction; they must never +be reported as executed evidence. +""" + +from __future__ import annotations + +import base64 +import os +from datetime import datetime, timezone + +import pytest + +from openadapt_flow.eligibility.artifact import ( + ArtifactEncryption, + PracticeArtifactPolicy, + all_confirmed, + write_and_verify, +) +from openadapt_flow.eligibility.client import ( + ApplicationMode, + BenefitSelection, + EligibilityRequest, + EligibilityStatus, + StediAccountBoundary, + StediEligibilityClient, +) + +pytestmark = pytest.mark.skipif( + not os.environ.get("STEDI_API_KEY"), + reason=( + "STEDI_API_KEY is absent: create a Stedi sandbox TEST key, then run " + "`pytest -q tests/test_eligibility_live_stedi.py -rs`; no live " + "eligibility evidence was collected" + ), +) + + +@pytest.mark.parametrize("trial", [1, 2, 3]) +def test_stedi_dental_mock_roundtrip_three_trials(tmp_path, trial): + account = StediAccountBoundary( + practice_account_id="openadapt-sandbox", + application_mode=ApplicationMode.TEST, + ) + client = StediEligibilityClient(account=account) + request = EligibilityRequest( + operation_id=f"stedi-cigna-dental-live-trial-{trial}", + payer_id="62308", + member_id="U3141592653", + first_name="Jaguar", + last_name="Dent", + date_of_birth="19960505", + provider_npi="1999999984", + provider_organization="One", + service_type_codes=["35"], + date_of_service=datetime.now(timezone.utc).strftime("%Y%m%d"), + benefit_selection=BenefitSelection(network_code="Y", coverage_level_code="IND"), + ) + result = client.check(request) + assert result.is_answer, result.reason + assert result.status in (EligibilityStatus.ACTIVE, EligibilityStatus.INACTIVE) + assert result.application_mode is ApplicationMode.TEST + assert result.raw_271_sha256 + + artifact_key = base64.urlsafe_b64encode(os.urandom(32)).decode() + policy = PracticeArtifactPolicy( + boundary_id=f"live-test-trial-{trial}", + application_mode=ApplicationMode.TEST, + encryption=ArtifactEncryption.APPLICATION_AES256_GCM, + encryption_key_env="LIVE_TEST_ARTIFACT_KEY", + retention_days=1, + ) + artifact, verdicts = write_and_verify( + result, + tmp_path, + request=request, + policy=policy, + env={"LIVE_TEST_ARTIFACT_KEY": artifact_key}, + ) + assert all_confirmed(verdicts), [v.reason for v in verdicts] + assert artifact.raw_271_file.endswith(".enc") diff --git a/tests/test_eligibility_waterfall.py b/tests/test_eligibility_waterfall.py new file mode 100644 index 00000000..4ab222d8 --- /dev/null +++ b/tests/test_eligibility_waterfall.py @@ -0,0 +1,299 @@ +"""Exact payer/account/service routing tests for the eligibility waterfall.""" + +from __future__ import annotations + +import httpx +import pytest + +from openadapt_flow.eligibility.client import ( + ApplicationMode, + EligibilityRequest, + EligibilityStatus, + StediAccountBoundary, + StediEligibilityClient, +) +from openadapt_flow.eligibility.waterfall import ( + DEFAULT_REGISTRY_PATH, + EligibilityRoute, + PayerCapability, + PayerRegistry, + load_payer_routes, + resolve_route, + run_waterfall, +) + +DIGEST = "a" * 64 +ENV = {"STEDI_API_KEY": "test-key"} +ACCOUNT = StediAccountBoundary( + practice_account_id="practice-1", application_mode=ApplicationMode.TEST +) + + +def request_for(payer_id: str = "62308", services=None) -> EligibilityRequest: + return EligibilityRequest( + operation_id="route-check-1", + payer_id=payer_id, + provider_npi="1999999984", + provider_organization="One", + member_id="123", + service_type_codes=services or ["35"], + date_of_service="20260721", + ) + + +def api_capability(**overrides) -> PayerCapability: + data = dict( + key="cigna", + display_name="Cigna Dental", + route=EligibilityRoute.API, + request_payer_id="62308", + stedi_id="HGJLR", + application_mode=ApplicationMode.TEST, + practice_account_id="practice-1", + supported_service_type_codes=["35"], + payer_record_sha256=DIGEST, + portal_fallback_reviewed=True, + verified_on="2026-07-21", + ) + data.update(overrides) + return PayerCapability(**data) + + +def registry(cap=None) -> PayerRegistry: + item = cap or api_capability() + return PayerRegistry(payers={item.key: item}) + + +def fake_client(body: dict, *, status=200, account=ACCOUNT, max_attempts=1): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(status, json=body) + + return StediEligibilityClient( + account=account, + transport=httpx.MockTransport(handler), + env=ENV, + max_attempts=max_attempts, + sleep=lambda _: None, + ) + + +ACTIVE = { + "meta": {"applicationMode": "test"}, + "tradingPartnerServiceId": "62308", + "subscriber": {"memberId": "123"}, + "provider": {"npi": "1999999984"}, + "planDateInformation": {"planBegin": "20260101", "planEnd": "20261231"}, + "benefitsInformation": [{"code": "1", "serviceTypeCodes": ["35"]}], +} + + +def test_shipped_registry_is_a_complete_synthetic_test_binding(): + assert DEFAULT_REGISTRY_PATH.exists() + loaded = load_payer_routes() + assert loaded.default_route is EligibilityRoute.QUEUE + cap = loaded.payers["cigna_dental_mock"] + assert cap.application_mode is ApplicationMode.TEST + assert cap.request_payer_id == "62308" + assert cap.stedi_id == "HGJLR" + assert cap.supported_service_type_codes == ["35"] + assert len(cap.payer_record_sha256 or "") == 64 + + +def test_incomplete_api_binding_fails_registry_validation(): + with pytest.raises(ValueError, match="exact reviewed binding"): + PayerCapability(key="unsafe", route=EligibilityRoute.API) + + +def test_unknown_payer_queues_instead_of_guessing_a_portal(): + decision = resolve_route("Unknown Dental", registry()) + assert decision.route is EligibilityRoute.QUEUE + assert "no exact reviewed route" in decision.reason + + +def test_exact_request_and_stedi_ids_resolve_the_same_reviewed_route(): + exact = resolve_route("62308", registry()) + stable = resolve_route("HGJLR", registry()) + assert exact.route is EligibilityRoute.API + assert stable.route is EligibilityRoute.API + assert exact.capability == stable.capability + + +def test_non_queue_default_is_refused(): + with pytest.raises(ValueError, match="attended queue"): + PayerRegistry(default_route=EligibilityRoute.PORTAL) + + +def test_alias_collision_fails_loud(): + with pytest.raises(ValueError, match="maps to both"): + PayerRegistry( + payers={ + "one": api_capability(key="one", aliases=["shared"]), + "two": api_capability( + key="two", + display_name="Other", + aliases=["shared"], + request_payer_id="99999", + ), + } + ) + + +def test_exact_binding_answer_completes_api_route(): + outcome = run_waterfall( + request_for(), + payer="Cigna Dental", + client=fake_client(ACTIVE), + registry=registry(), + ) + assert outcome.final_route is EligibilityRoute.API + assert outcome.answered_by_api + + +def test_request_payer_id_mismatch_never_calls_api(): + calls = 0 + + def handler(request: httpx.Request) -> httpx.Response: + nonlocal calls + calls += 1 + return httpx.Response(200, json=ACTIVE) + + client = StediEligibilityClient( + account=ACCOUNT, transport=httpx.MockTransport(handler), env=ENV + ) + outcome = run_waterfall( + request_for("6238"), payer="Cigna Dental", client=client, registry=registry() + ) + assert calls == 0 + assert outcome.final_route is EligibilityRoute.QUEUE + assert "payer ID" in outcome.trail[-1] + + +def test_unreviewed_service_type_never_calls_api(): + outcome = run_waterfall( + request_for(services=["30"]), + payer="Cigna Dental", + client=fake_client(ACTIVE), + registry=registry(), + ) + assert outcome.final_route is EligibilityRoute.QUEUE + assert "service type" in outcome.trail[-1] + + +def test_account_or_mode_mismatch_never_calls_api(): + other = StediAccountBoundary( + practice_account_id="practice-2", application_mode=ApplicationMode.TEST + ) + outcome = run_waterfall( + request_for(), + payer="Cigna Dental", + client=fake_client(ACTIVE, account=other), + registry=registry(), + ) + assert outcome.final_route is EligibilityRoute.QUEUE + assert "account/mode" in outcome.trail[-1] + + +def test_transient_failure_falls_to_reviewed_portal_after_retries(): + body = {"code": "TOO_MANY_REQUESTS"} + outcome = run_waterfall( + request_for(), + payer="Cigna Dental", + client=fake_client(body, status=429, max_attempts=2), + registry=registry(), + ) + assert outcome.final_route is EligibilityRoute.PORTAL + assert outcome.result is not None + assert outcome.result.status is EligibilityStatus.PAYER_UNAVAILABLE + assert outcome.result.attempt_count == 2 + + +def test_member_identity_error_queues_without_portal_fallback(): + body = { + "meta": {"applicationMode": "test"}, + "tradingPartnerServiceId": "62308", + "errors": [{"code": "75"}], + } + outcome = run_waterfall( + request_for(), + payer="Cigna Dental", + client=fake_client(body), + registry=registry(), + ) + assert outcome.final_route is EligibilityRoute.QUEUE + assert outcome.result is not None + assert outcome.result.status is EligibilityStatus.NOT_FOUND + + +def test_portal_banned_transient_failure_queues(): + cap = api_capability(portal_banned=True, portal_fallback_reviewed=False) + outcome = run_waterfall( + request_for(), + payer="Cigna Dental", + client=fake_client({"error": "down"}, status=503), + registry=registry(cap), + ) + assert outcome.final_route is EligibilityRoute.QUEUE + assert "barred" in outcome.trail[-1] + + +def test_transient_failure_without_reviewed_portal_fallback_queues(): + cap = api_capability(portal_fallback_reviewed=False) + outcome = run_waterfall( + request_for(), + payer="Cigna Dental", + client=fake_client({"error": "down"}, status=503), + registry=registry(cap), + ) + assert outcome.final_route is EligibilityRoute.QUEUE + assert "no portal fallback is explicitly reviewed" in outcome.trail[-1] + + +def test_explicit_portal_route_needs_no_api_key(): + cap = PayerCapability( + key="portal", display_name="Portal Dental", route=EligibilityRoute.PORTAL + ) + outcome = run_waterfall( + request_for("PORTAL"), payer="Portal Dental", registry=registry(cap) + ) + assert outcome.final_route is EligibilityRoute.PORTAL + assert outcome.result is None + + +def test_portal_banned_configuration_cannot_select_portal(): + with pytest.raises(ValueError, match="portal-banned"): + PayerCapability( + key="contradictory", + display_name="Contradictory Portal", + route=EligibilityRoute.PORTAL, + portal_banned=True, + ) + + +def test_excluded_route_stays_excluded(): + cap = PayerCapability( + key="blocked", + display_name="Blocked Portal", + route=EligibilityRoute.EXCLUDED, + portal_banned=True, + ) + outcome = run_waterfall( + request_for("BLOCKED"), payer="Blocked Portal", registry=registry(cap) + ) + assert outcome.final_route is EligibilityRoute.EXCLUDED + + +def test_malformed_registry_fails_loud(tmp_path): + path = tmp_path / "routes.yaml" + path.write_text("just a string") + with pytest.raises(ValueError, match="payer mapping"): + load_payer_routes(path) + + +def test_lazy_package_exports(): + import openadapt_flow.eligibility as eligibility + + assert eligibility.EligibilityRoute is EligibilityRoute + assert callable(eligibility.run_waterfall) + assert callable(eligibility.purge_expired_eligibility_artifacts) + with pytest.raises(AttributeError): + eligibility.nope