Configuring TSH Reflex Cascades in a Python Rule Engine

Problem Statement

Thyroid testing is the archetypal reflex cascade, and it is where naïve rule engines quietly misbehave. The standard protocol runs TSH first and, only when the TSH falls outside its reference interval, reflexes to free T4 — an abnormal TSH with a normal free T4 points to subclinical disease, a suppressed TSH with a high free T4 confirms hyperthyroidism. Some laboratories extend the cascade a step further: a suppressed TSH reflexes to free T3 to characterize T3-toxicosis, and a persistently elevated TSH with normal free T4 reflexes to thyroid peroxidase (TPO) antibodies to work up autoimmune hypothyroidism. Each of those “onward” steps depends on the result of the previous reflex, not just the original TSH, so the cascade is genuinely ordered — and an engine that fires all rules against the original TSH in a single flat pass either orders the wrong second-tier test or floods the order queue. This page implements the thyroid cascade as ordered, depth-tracked conditional rules, with the exact Pydantic v2 models and audit records that make each firing reproducible.

Prerequisites

Before wiring the cascade into the validation path, confirm the following baseline:

  • Runtime: Python 3.11+ (for match statements and datetime UTC helpers), pydantic>=2.6, and pytest>=8 with pytest-asyncio for the verification fixtures.
  • Upstream classification: each thyroid result arrives already classified by the Reference Range Check Implementation, carrying a flag and a Decimal value in canonical units (TSH in m[IU]/L, free T4 in pmol/L), with analyte identity resolved to LOINC.
  • Routing host: the cascade runs inside the Reflex Testing Routing Logic node, which supplies the ReflexRule and ReflexOrder contracts and the specimen reflex_lineage this page builds on.
  • Governance baseline: the reflex algorithm is documented and approved by the laboratory’s medical director, and an append-only audit sink is available for the CLIA retention window.

Step-by-Step Implementation

Step 1: Model the cascade as ordered conditional rules

The thyroid cascade is three rules, and their order is load-bearing. Each rule names the analyte that must have just resulted to arm it (trigger_loinc), the condition on that result, and the target to order. The first rule triggers on TSH; the second and third trigger on free T4, which only exists because the first rule fired. Model them so the engine can fetch candidates by the analyte that actually resulted, rather than replaying the whole cascade against TSH.

python
from __future__ import annotations

from decimal import Decimal

from pydantic import BaseModel, ConfigDict, Field

TSH = "3016-3"       # Thyrotropin (TSH), serum
FT4 = "3024-7"       # Free thyroxine (FT4), serum
FT3 = "3051-0"       # Free triiodothyronine (FT3), serum
TPO_AB = "8099-8"    # Thyroid peroxidase antibody, serum


class ThyroidRule(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")

    rule_id: str
    trigger_loinc: str          # analyte that must have just resulted
    trigger_flag_in: frozenset[str]  # arming flags on that analyte
    require_prior_normal: str | None = None  # a target that must be NORMAL to fire
    target_loinc: str
    target_test_code: str
    max_cascade_depth: int = Field(ge=1, default=3)
    version: str


CASCADE: tuple[ThyroidRule, ...] = (
    ThyroidRule(
        rule_id="thy-tsh-ft4", trigger_loinc=TSH,
        trigger_flag_in=frozenset({"HIGH", "LOW", "CRITICAL_HIGH", "CRITICAL_LOW"}),
        target_loinc=FT4, target_test_code="FT4", version="thy-v4",
    ),
    ThyroidRule(
        rule_id="thy-ft4-ft3", trigger_loinc=FT4,
        trigger_flag_in=frozenset({"NORMAL", "HIGH"}),
        target_loinc=FT3, target_test_code="FT3", version="thy-v4",
    ),
    ThyroidRule(
        rule_id="thy-ft4-tpo", trigger_loinc=FT4,
        trigger_flag_in=frozenset({"NORMAL"}),
        target_loinc=TPO_AB, target_test_code="TPOAB", version="thy-v4",
    ),
)

The require_prior_normal field is deliberately present but unset here; it exists so a laboratory can express “reflex TPO antibodies only when free T4 came back normal and the original TSH was elevated,” a condition Step 3 evaluates against the specimen’s accumulated results rather than a single value.

Step 2: Evaluate the primary TSH and emit the first reflex

The engine keys candidate rules on the analyte that resulted. When the TSH result arrives, only thy-tsh-ft4 is a candidate; it fires when the TSH flag is abnormal, emitting a free T4 order at cascade depth one. Comparisons stay on the flag the range check already assigned, so the cascade never re-derives the reference interval.

python
from enum import Enum


class Flag(str, Enum):
    NORMAL = "NORMAL"
    LOW = "LOW"
    HIGH = "HIGH"
    CRITICAL_LOW = "CRITICAL_LOW"
    CRITICAL_HIGH = "CRITICAL_HIGH"


class ThyroidResult(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")

    loinc: str
    value: Decimal
    unit: str
    specimen_id: str
    flag: Flag
    reflex_lineage: tuple[str, ...] = ()
    cascade_depth: int = 0


class ReflexOrder(BaseModel):
    model_config = ConfigDict(frozen=True, extra="forbid")

    rule_id: str
    rule_version: str
    source_loinc: str
    target_loinc: str
    target_test_code: str
    specimen_id: str
    cascade_depth: int


def fire(rule: ThyroidRule, result: ThyroidResult) -> ReflexOrder:
    return ReflexOrder(
        rule_id=rule.rule_id, rule_version=rule.version,
        source_loinc=result.loinc, target_loinc=rule.target_loinc,
        target_test_code=rule.target_test_code, specimen_id=result.specimen_id,
        cascade_depth=result.cascade_depth + 1,
    )

Step 3: Extend the cascade with depth-tracked conditional rules

When the free T4 result arrives, it carries cascade_depth = 1 and a reflex_lineage that already contains the free T4 LOINC. Now two rules are candidates. To evaluate the “onward” conditions correctly the engine needs the original TSH direction, so the cascade evaluator receives a small specimen_context mapping of the analytes already resulted on this specimen. Depth is checked on every firing so a mis-configured rule cannot loop.

python
def evaluate_cascade(
    result: ThyroidResult,
    context: dict[str, Flag],
    rules: tuple[ThyroidRule, ...] = CASCADE,
) -> tuple[list[ReflexOrder], list[tuple[str, str]]]:
    emitted: list[ReflexOrder] = []
    suppressed: list[tuple[str, str]] = []
    for rule in rules:
        if rule.trigger_loinc != result.loinc:
            continue
        if result.flag.value not in rule.trigger_flag_in:
            continue
        if rule.target_loinc in result.reflex_lineage:
            suppressed.append((rule.rule_id, "already_reflexed"))
            continue
        if result.cascade_depth + 1 > rule.max_cascade_depth:
            suppressed.append((rule.rule_id, "max_cascade_depth"))
            continue
        # onward TPO rule fires only when the original TSH was elevated
        if rule.rule_id == "thy-ft4-tpo":
            tsh_flag = context.get(TSH)
            if tsh_flag not in (Flag.HIGH, Flag.CRITICAL_HIGH):
                suppressed.append((rule.rule_id, "tsh_not_elevated"))
                continue
        emitted.append(fire(rule, result))
    return emitted, suppressed

The TPO branch shows why the cascade must be ordered: it fires on a normal free T4, but only when the original TSH was elevated — the exact subclinical-hypothyroid pattern. Evaluating it against the TSH alone would be impossible, because free T4 has not resulted yet; evaluating it in a flat pass would fire it whenever any free T4 is normal, ordering TPO antibodies on euthyroid patients. Keying on the resulted analyte and consulting the specimen context is what keeps each step clinically correct.

Step 4: Emit an immutable audit record for every firing

Every cascade evaluation — fired or suppressed — produces a replayable record so an inspector can reconstruct why a given specimen did or did not reflex. The audit captures the resulted analyte, its flag, the rule version, and the suppression reason where applicable.

python
import datetime as dt


class CascadeAudit(BaseModel):
    trace_id: str
    specimen_id: str
    resulted_loinc: str
    resulted_flag: Flag
    emitted: list[str]           # target LOINCs ordered
    suppressed: list[tuple[str, str]]
    rule_version: str
    evaluated_at: dt.datetime = Field(
        default_factory=lambda: dt.datetime.now(dt.timezone.utc)
    )
TSH reflex cascade A TSH result is evaluated. When the TSH is within its reference interval no reflex fires. When the TSH is abnormal the engine orders a free T4 at cascade depth one. When the free T4 then results, a suppressed original TSH reflexes to free T3, and an elevated original TSH with a normal free T4 reflexes to TPO antibodies, both at cascade depth two. Every emitted order and the free T4 order itself are written to the audit sink. TSH result Evaluate TSH condition flag_in HIGH or LOW abnormal within range Order Free T4 reflex order · depth 1 No reflex TSH normal TSH low TSH high, FT4 normal Order Free T3 depth 2 Order TPO antibodies depth 2 Reflex orders emitted + audited

Verification & Testing

Prove that the cascade fires in order and that the onward conditions honour the original TSH. pytest fixtures pin the three clinically distinct scenarios — a normal TSH that reflexes nothing, a suppressed TSH that reflexes free T4 then free T3, and an elevated TSH with a normal free T4 that reflexes TPO antibodies — plus the negative case where a normal free T4 on a low TSH does not order TPO.

python
import pytest


def _tsh(flag: Flag) -> ThyroidResult:
    return ThyroidResult(loinc=TSH, value=Decimal("8.2"), unit="m[IU]/L",
                         specimen_id="S1", flag=flag)


def test_normal_tsh_reflexes_nothing() -> None:
    emitted, _ = evaluate_cascade(_tsh(Flag.NORMAL), context={TSH: Flag.NORMAL})
    assert emitted == []


def test_abnormal_tsh_orders_ft4() -> None:
    emitted, _ = evaluate_cascade(_tsh(Flag.HIGH), context={TSH: Flag.HIGH})
    assert [o.target_loinc for o in emitted] == [FT4]
    assert emitted[0].cascade_depth == 1


def test_normal_ft4_after_high_tsh_orders_tpo() -> None:
    ft4 = ThyroidResult(loinc=FT4, value=Decimal("14.0"), unit="pmol/L",
                        specimen_id="S1", flag=Flag.NORMAL,
                        reflex_lineage=(FT4,), cascade_depth=1)
    emitted, suppressed = evaluate_cascade(ft4, context={TSH: Flag.HIGH})
    assert TPO_AB in [o.target_loinc for o in emitted]
    assert all(o.cascade_depth == 2 for o in emitted)


def test_normal_ft4_after_low_tsh_suppresses_tpo() -> None:
    ft4 = ThyroidResult(loinc=FT4, value=Decimal("15.0"), unit="pmol/L",
                        specimen_id="S1", flag=Flag.NORMAL,
                        reflex_lineage=(FT4,), cascade_depth=1)
    _, suppressed = evaluate_cascade(ft4, context={TSH: Flag.LOW})
    assert ("thy-ft4-tpo", "tsh_not_elevated") in suppressed

Expected: all four pass. test_normal_ft4_after_high_tsh_orders_tpo confirms the ordered, context-aware onward reflex; test_normal_ft4_after_low_tsh_suppresses_tpo confirms the negative — a normal free T4 on a suppressed TSH does not fabricate a TPO order. Round-trip CascadeAudit through model_dump_json() / model_validate_json() in a fifth test so the record written to the sink deserializes back identically for later replay.

Compliance Note

CLIA §493.1253 requires that the reflex algorithm be an established, verified procedure; storing each ThyroidRule with a version and pinning the cascade in a version-controlled configuration makes the exact algorithm that fired reconstructable at inspection. The medical-director-approved reflex protocol — the clinical decision that an abnormal TSH warrants a free T4 and, conditionally, TPO antibodies — is captured in the rule set and its approval metadata, satisfying the CAP requirement that automated reflex logic be documented and reviewable. Emitting a CascadeAudit for every evaluation, fired or suppressed, and writing it to an append-only sink under the CLIA §493.1105 retention window satisfies the 21 CFR Part 11 expectation that an electronic ordering decision be attributable and reconstructable years later. Because the cascade orders tests on a patient without a physician re-touching the chart, the audit must be written before the order is published.

Troubleshooting

TPO antibodies are ordered on patients with a suppressed TSH.

Root cause: the onward TPO rule was evaluated against the free T4 result alone, so any normal free T4 armed it. Fix: consult the specimen context for the original TSH direction, as evaluate_cascade does, and suppress thy-ft4-tpo unless context[TSH] is HIGH or CRITICAL_HIGH. The test_normal_ft4_after_low_tsh_suppresses_tpo fixture pins the negative case.

Free T4 is ordered twice on the same specimen.

Root cause: the free T4 result re-entered the cascade and re-matched the TSH rule, or the order was re-driven after a transient failure. Fix: the free T4 LOINC is added to reflex_lineage when its order is emitted, so the already_reflexed guard suppresses a second firing; ensure the downstream ordering service also deduplicates idempotently on rule_id, specimen_id, and target_loinc.

The onward free T3 or TPO reflex never fires.

Root cause: the free T4 result arrived without its cascade_depth and reflex_lineage propagated, so the engine treated it as a fresh depth-zero order and never matched the free-T4-triggered rules — or the specimen context was empty, so the TPO condition could not see the original TSH. Fix: propagate cascade_depth and reflex_lineage onto every reflex-generated result and populate the context with each analyte’s flag as it results.

A mis-configured pair of thyroid rules floods the order queue.

Root cause: two rules point at each other’s trigger analyte with no depth bound, so each emission arms the next indefinitely. Fix: every ThyroidRule carries a max_cascade_depth, and the evaluator suppresses any firing that would exceed it with a max_cascade_depth reason. Add the no-cycle assertion from the routing node’s rule-set contract test to CI so the pair is rejected before promotion.

Part of: Reflex Testing Routing Logic.