feat(bcda): rewrite the input_layer connector for v3 semantics

The connector (bcda.express.cclf / bcda.pipe.cclf) still routed on the v2
near-line RIC code, read v2-only amount columns, and had never run end to
end: its contracts named CCLF columns the EOB table does not have, so the
runner's exact-match check would have refused every stage.

One header stage now carries the v3 semantics. _stg_claim_header drops
NotFinalAction claims (9% of the sandbox), derives claim_family from
EOB-TYPE → CLM-TYPE-CD → pde_id → RIC (v3 emits the RIC on only some
claims), collapses a claim seen in two sources on its claim control number
(NationalClaimsHistory/DDPS over SharedSystem, then newest last_updated;
rows without a control number are never collapsed) and resolves the MBI.
The four family stages are filters on it.

One mapping produces input_layer.medical_claim for institutional, carrier
and DME lines: admission/discharge from the C4BB admission period, NUBC
type of bill (new clm_bill_type_cd on the EOB table; rebuilt as "0" + the
three-character v3 frequency code when the coding is missing — the v2
fac+clsfctn+freq concatenation yields "88787J" on v3), MS-DRG, revenue
centre, HCPCS + modifiers, attending→operating→rendering / rendering /
rendering→ordering NPIs, billing Organization NPI and tax number,
paid/allowed/charge/coinsurance/deductible from the consolidated v3
Adjudication amounts, paid date from clm_pd_dt → NCH weekly → CMS process.
Part D uses the PDE id, prescribing care-team NPI, contained pharmacy NPI,
NDC quantity, total drug cost as charge and paid-by-patient as copayment.
Eligibility reads the Part A row (Part B fallback) for status, entitlement
and enrollment, the Dual row for dual status, and accepts both CCW sex
codes and the FHIR words the v2 flattener passed through.

Contracts inherit: header stages subclass ExplanationOfBenefit, outputs
subclass aco.table.input_layer.InputLayer*, so downstream pipelines run on
BCDA output exactly as on CCLF output. load_bcda(run_pipeline=True),
stack load bcda --run-pipeline and stack run bcda land input_layer.*.

Large-entity sandbox pull: 221,021 medical lines on 28,282 claims (equal to
the FinalAction non-PDE EOBs by type), 11,968 pharmacy rows, 9,582
eligibility rows; all 16 reconciliation checks in the notebook pass. Tests
run the whole connector on CMS's official v3 samples through DuckDB.

refs #1139

docs/docs/cli/load-bcda.md

@@ -19,6 +19,8 @@ Usage: stack load bcda [OPTIONS]
 │ --ndjson-dir          PATH  Directory containing FHIR ndjson files.          │
 │ --database            TEXT  DuckDB database path (default: config).          │
 │ --skip-flatten              Skip flattening, load existing Parquet.          │
+│ --run-pipeline              Also run the bcda connector pipeline into        │
+│                             input_layer.*                                    │
 │ --help                      Show this message and exit.                      │
 ╰──────────────────────────────────────────────────────────────────────────────╯
 ```

docs/superpowers/specs/2026-10-25-bcda-v3-design.md

@@ -52,3 +52,15 @@ Pulled as three single-type jobs: the EOB job (89596, 6 min server-side, 192 fil
 | `bcda.eob_item` | 248,528 |
 
 All 20 notebook checks pass: every Patient has a unique current MBI and `bene_id == Patient.id`; every Coverage row has `period.start`, every Part A row `ms_cd`, every Dual row `dual_stus_cd`, no orphan beneficiaries; every EOB has a claim id, a claim control number, System-Type and Final-Action tags, EOB-TYPE and CLM-TYPE-CD; control numbers are unique within a source (v3 deduplication holds); lines carry HCPCS or NDC on 100% and paid/submitted amounts on 94% (Part D lines carry drug-cost amounts instead); no orphan lines. The entity is adjudicated-only (no SharedSystem claims), so the control-number pairing check is exercised but finds zero pairs — partially adjudicated data needs an entity with PAC access and `_typeFilter`.
+
+## Connector pipeline on v3 semantics (`bcda.express.cclf`, 2026-10-25)
+
+The input_layer connector was still written for v2 and had never run end to end (its contracts named CCLF columns the EOB table does not have, and the runner's exact-match check would have refused every stage). Rewritten:
+
+1. **One header stage carries the v3 semantics.** `_stg_claim_header` drops `NotFinalAction` claims, derives `claim_family` from `EOB-TYPE` → `CLM-TYPE-CD` → `pde_id` → the v2 near-line RIC (which v3 emits on only some claims: HHA shows U/W/V/NULL), collapses a claim seen in two sources on its claim control number (NationalClaimsHistory/DDPS over SharedSystem, then newest `last_updated`; rows without a control number are never collapsed), and resolves the MBI. The four family stages are filters on it.
+2. **One mapping for medical_claim.** `_medical_claim(kind)` produces the full `aco.table.input_layer.InputLayerMedicalClaim` shape for institutional, carrier and DME lines: admission/discharge from the C4BB admission period, NUBC type of bill (`clm_bill_type_cd`, rebuilt as `0` + the three-character v3 frequency code when the coding is missing), MS-DRG, revenue centre, HCPCS and five modifiers, attending→operating→rendering / rendering / rendering→ordering NPIs, billing Organization NPI and tax number, paid / allowed / charge / coinsurance / deductible from the consolidated v3 Adjudication amounts, paid date from `clm_pd_dt` → NCH weekly process → CMS process date, 12 header diagnoses with POA on the first, first ICD procedure.
+3. **Part D** uses the PDE id, the prescribing care-team NPI, the contained pharmacy Organization NPI, NDC quantity, total drug cost as charge and paid-by-patient as copayment; plan-paid falls back line → header (the sandbox emits neither).
+4. **Eligibility** reads the Part A row (Part B fallback) for status, entitlement and the enrollment period and the Dual row for dual status; gender accepts CCW codes and the FHIR words the v2 flattener passed through; `enrollment_status` comes from `Coverage.status`.
+5. **Contracts inherit.** The header stages subclass `ExplanationOfBenefit`; the outputs subclass the canonical `InputLayer*` models, so the same downstream pipelines run on BCDA output as on CCLF output. `load_bcda(run_pipeline=True)` / `stack load bcda --run-pipeline` and `stack run bcda` land `input_layer.medical_claim`, `pharmacy_claim`, `eligibility`.
+
+Run on the Large-entity pull (94 s including re-flattening): 221,021 medical lines on 28,282 claims (16,468 institutional, 11,814 professional — equal to the FinalAction non-PDE EOB counts by type; 4,111 NotFinalAction excluded), 11,968 pharmacy rows (= FinalAction PDEs), 9,582 eligibility rows (7,290 with a Part A/B row, 4,868 with dual status). Every institutional line has a 4-character type of bill, admission date, paid, charge, deductible and coinsurance; every professional line has paid, rendering NPI and (99 %) place of service.

notebooks/bcda_v3_sandbox.py

@@ -30,11 +30,15 @@ def _(mo):
        younger than 24 h is **adopted** instead of re-run.
     4. **Flatten + load** — `bcda.express.v3` (table-driven from the v3 Data
        Dictionary's `fhirPath` / `sourceColumn` / `ccwMapping` columns) streams
-       the ndjson into Parquet and `aco.duckdb` `bcda.*`, then the read-only
-       replica is republished.
+       the ndjson into Parquet and `aco.duckdb` `bcda.*`, runs the connector
+       pipeline (`bcda.pipe.cclf`: v3 routing on EOB-TYPE, Final-Action filter,
+       SharedSystem→NCH dedupe) into `input_layer.*`, then republishes the
+       read-only replica.
     5. **Proof** — identity, coverage shape, claim provenance, line amounts,
        referential integrity, and how many dictionary-mapped columns actually
        carry data, each marked pass / fail.
+    6. **input_layer** — the connector's medical_claim / pharmacy_claim /
+       eligibility reconciled against the `bcda.*` tables they came from.
 
     Network steps run only when **`BCDA_PULL=1`** is set or the buttons below are
     pressed (`BCDA_LOAD=1` flattens and loads what is already in the store without
@@ -338,8 +342,8 @@ def _(LOAD, PULL, STORE_PATH, connect, load_btn, mo, pulled, store, time):
     if _do_load:
         _dirs = _find_latest_export(STORE_PATH)
         _t0 = time.time()
-        load_stats = (
-            load_bcda()
+        load_stats = load_bcda(
+            run_pipeline=True
         )  # newest export per resource type → bcda.* in aco.duckdb
         _dt = time.time() - _t0
         _replica = connect.publish_replica("aco")
@@ -641,9 +645,187 @@ def _(alt, mo, plans, prov, q):
     return
 
 
+@app.cell(hide_code=True)
+def _(mo, q):
+    # ── 6. input_layer — the connector pipeline reconciled against bcda.* ──
+    _il = q(
+        "select table_name from information_schema.tables where table_schema = 'input_layer'"
+    )["table_name"].to_list()
+    mo.stop(
+        not {"medical_claim", "pharmacy_claim", "eligibility"} <= set(_il),
+        mo.md(
+            "## 6 · input_layer\n\n⚠️ `input_layer.*` not in the replica yet — run the "
+            "load above (it runs `bcda.pipe.cclf`) or `stack load bcda --skip-flatten --run-pipeline`."
+        ),
+    )
+    il_checks: list[tuple[bool, str, str]] = []
+
+    def il(ok: bool, name: str, detail: str) -> None:
+        il_checks.append((bool(ok), name, detail))
+
+    _src = q("""
+        select
+            count(*) filter (where final_action = 'FinalAction' and eob_type <> 'PDE') final_medical,
+            count(*) filter (where final_action = 'FinalAction' and eob_type = 'PDE') final_pde,
+            count(*) filter (where final_action = 'NotFinalAction') not_final,
+            count(*) filter (where final_action = 'FinalAction' and eob_type in ('INPATIENT','OUTPATIENT','SNF','HHA','HOSPICE')) final_inst,
+            count(*) filter (where final_action = 'FinalAction' and eob_type in ('CARRIER','DME')) final_prof
+        from bcda.explanation_of_benefit
+    """).to_dicts()[0]
+    _mc = q("""
+        select claim_type, count(*) lines, count(distinct claim_id) claims,
+               count(person_id) with_person, count(paid_amount) with_paid, count(charge_amount) with_charge,
+               count(deductible_amount) with_ded, count(coinsurance_amount) with_coins,
+               count(hcpcs_code) with_hcpcs, count(rendering_npi) with_rendering, count(billing_npi) with_billing,
+               count(bill_type_code) with_tob, count(drg_code) with_drg, count(admission_date) with_adm,
+               count(place_of_service_code) with_pos, count(diagnosis_code_1) with_dx, count(paid_date) with_paid_dt
+        from input_layer.medical_claim group by 1 order by 1
+    """)
+    _by = {r["claim_type"]: r for r in _mc.to_dicts()}
+    _inst, _prof = _by.get("institutional", {}), _by.get("professional", {})
+    _claims = sum(r["claims"] for r in _mc.to_dicts())
+    il(
+        _claims == _src["final_medical"],
+        "Every final-action medical EOB is in medical_claim, once",
+        f"{_claims:,} claims = {_src['final_medical']:,} FinalAction non-PDE EOBs; {_src['not_final']:,} NotFinalAction excluded",
+    )
+    il(
+        _inst.get("claims") == _src["final_inst"]
+        and _prof.get("claims") == _src["final_prof"],
+        "Routed by EOB-TYPE: institutional vs professional",
+        f"institutional {_inst.get('claims', 0):,} / {_src['final_inst']:,} · professional {_prof.get('claims', 0):,} / {_src['final_prof']:,}",
+    )
+    _no_person = q(
+        "select count(*) n from input_layer.medical_claim where person_id is null"
+    ).to_dicts()[0]["n"]
+    il(
+        _no_person == 0,
+        "Every line resolves to an MBI (person_id)",
+        f"{_no_person:,} lines without MBI",
+    )
+    il(
+        _inst.get("with_adm") == _inst.get("lines")
+        and _inst.get("with_tob", 0) >= 0.99 * _inst.get("lines", 1),
+        "Institutional lines carry admission date and NUBC type of bill",
+        f"TOB {_inst.get('with_tob', 0):,}, admission {_inst.get('with_adm', 0):,} of {_inst.get('lines', 0):,}",
+    )
+    _tob_len = q(
+        "select count(*) n from input_layer.medical_claim where claim_type='institutional' and length(bill_type_code) <> 4"
+    ).to_dicts()[0]["n"]
+    il(
+        _tob_len == 0,
+        "Type of bill is the 4-character NUBC code (not fac+clsfctn+freq)",
+        f"{_tob_len:,} malformed",
+    )
+    il(
+        _inst.get("with_paid") == _inst.get("lines")
+        and _inst.get("with_charge") == _inst.get("lines")
+        and _inst.get("with_ded") == _inst.get("lines")
+        and _inst.get("with_coins") == _inst.get("lines"),
+        "Institutional paid / charge / deductible / coinsurance from v3 Adjudication",
+        f"paid {_inst.get('with_paid', 0):,} · charge {_inst.get('with_charge', 0):,} · ded {_inst.get('with_ded', 0):,} · coins {_inst.get('with_coins', 0):,}",
+    )
+    il(
+        _prof.get("with_paid") == _prof.get("lines")
+        and _prof.get("with_pos", 0) >= 0.95 * _prof.get("lines", 1)
+        and _prof.get("with_rendering") == _prof.get("lines"),
+        "Professional lines: paid, place of service, rendering NPI",
+        f"paid {_prof.get('with_paid', 0):,} · POS {_prof.get('with_pos', 0):,} · rendering {_prof.get('with_rendering', 0):,} of {_prof.get('lines', 0):,}",
+    )
+    il(
+        _inst.get("with_drg", 0) > 0 and _inst.get("with_dx", 0) == _inst.get("lines"),
+        "Institutional DRG and header diagnoses present",
+        f"DRG {_inst.get('with_drg', 0):,} · dx1 {_inst.get('with_dx', 0):,}",
+    )
+    il(
+        (_inst.get("with_paid_dt", 0) + _prof.get("with_paid_dt", 0))
+        == (_inst.get("lines", 0) + _prof.get("lines", 0)),
+        "Paid date resolved (clm_pd_dt → NCH weekly → CMS process)",
+        f"{_inst.get('with_paid_dt', 0) + _prof.get('with_paid_dt', 0):,} lines",
+    )
+
+    _rx = q("""
+        select count(*) n, count(distinct claim_id) claims, count(prescribing_provider_npi) prs, count(dispensing_provider_npi) dsp,
+               count(ndc_code) ndc, count(quantity) qty, count(days_supply) ds, count(refills) rf, count(charge_amount) chg,
+               count(copayment_amount) cop, count(paid_amount) paid, count(paid_date) pd
+        from input_layer.pharmacy_claim
+    """).to_dicts()[0]
+    il(
+        _rx["claims"] == _src["final_pde"],
+        "Every final-action PDE is in pharmacy_claim, once",
+        f"{_rx['claims']:,} = {_src['final_pde']:,}",
+    )
+    il(
+        _rx["prs"] == _rx["n"]
+        and _rx["dsp"] == _rx["n"]
+        and _rx["ndc"] == _rx["n"]
+        and _rx["qty"] == _rx["n"]
+        and _rx["ds"] == _rx["n"],
+        "Part D: prescriber, pharmacy, NDC, quantity, days supply populated",
+        f"{_rx['n']:,} rows",
+    )
+    il(
+        _rx["chg"] == _rx["n"] and _rx["cop"] == _rx["n"],
+        "Part D: total drug cost and paid-by-patient populated",
+        f"charge {_rx['chg']:,} · copay {_rx['cop']:,} · plan-paid {_rx['paid']:,} (sandbox emits none)",
+    )
+
+    _el = q("""
+        select count(*) n, count(distinct person_id) people, count(gender) g, count(enrollment_start_date) es,
+               count(medicare_status_code) ms, count(original_reason_entitlement_code) orec, count(dual_status_code) dual,
+               count(enrollment_status) st, count(*) filter (where gender = 'unknown') g_unk, count(*) filter (where race = 'unknown') r_unk
+        from input_layer.eligibility
+    """).to_dicts()[0]
+    _pat = q("select count(*) n from bcda.patient").to_dicts()[0]["n"]
+    _dual_rows = q(
+        "select count(distinct subscriber_id) n from bcda.coverage where class_plan = 'Dual Medicare/Medicaid'"
+    ).to_dicts()[0]["n"]
+    _ab_rows = q(
+        "select count(distinct subscriber_id) n from bcda.coverage where class_plan in ('Part A', 'Part B')"
+    ).to_dicts()[0]["n"]
+    il(
+        _el["n"] == _pat and _el["people"] == _pat,
+        "One eligibility row per Patient",
+        f"{_el['n']:,} = {_pat:,}",
+    )
+    il(
+        _el["g_unk"] == 0,
+        "Gender resolved for every row",
+        f"{_el['g_unk']:,} unknown; race unknown {_el['r_unk']:,} (OMB UNK)",
+    )
+    il(
+        _el["es"] == _el["ms"] == _el["st"] == _ab_rows,
+        "Enrollment period, Medicare status and status from the Part A/B Coverage row",
+        f"{_el['es']:,} rows = {_ab_rows:,} beneficiaries with a Part A/B row; {_el['n'] - _el['es']:,} patients have none in this pull",
+    )
+    il(
+        _el["dual"] == _dual_rows,
+        "Dual status from the Dual Medicare/Medicaid Coverage row",
+        f"{_el['dual']:,} = {_dual_rows:,} beneficiaries with a Dual row",
+    )
+
+    _passed = sum(1 for ok, _, _ in il_checks if ok)
+    mo.vstack(
+        [
+            mo.md(f"## 6 · input_layer — {_passed} / {len(il_checks)} checks pass"),
+            mo.md(
+                "\n".join(
+                    f"- {'✅' if ok else '❌'} **{name}** — {detail}"
+                    for ok, name, detail in il_checks
+                )
+            ),
+            mo.ui.table(
+                _mc,
+                label="medical_claim by claim_type: lines, claims and populated fields",
+            ),
+        ]
+    )
+    return (il_checks,)
+
+
 @app.cell(hide_code=True)
 def _(ROOT, con, mo, pl, q):
-    # ── 6. Dictionary field population + CCLF crosswalk ──
+    # ── 7. Dictionary field population + CCLF crosswalk ──
     _cols = q(
         "select table_name, column_name from information_schema.columns where table_schema = 'bcda' order by 1, 2"
     )
@@ -721,7 +903,7 @@ def _(ROOT, con, mo, pl, q):
     mo.vstack(
         [
             mo.md(f"""
-            ## 6 · Dictionary field population and the CCLF crosswalk
+            ## 7 · Dictionary field population and the CCLF crosswalk
 
             Columns are named after the v3 Data Dictionary's `ccwMapping` (legacy CCW names) or its IDR
             `sourceColumn` (v3-only). A populated column means the v3 payload actually carried that element

src/aco/load/bcda.py

@@ -22,6 +22,7 @@ def load_bcda(
     database: str | None = None,
     *,
     skip_flatten: bool = False,
+    run_pipeline: bool = False,
 ) -> dict[str, int]:
     """Flatten BCDA ndjson and load into DuckDB.
 
@@ -37,6 +38,12 @@ def load_bcda(
         DuckDB database path.  Defaults to ``conf.path("db.aco")``.
     skip_flatten : bool
         If True, skip the flatten step and load existing Parquet files.
+    run_pipeline : bool
+        After loading, run the ``bcda`` connector pipeline
+        (``bcda.pipe.cclf``) and save ``input_layer.medical_claim``,
+        ``input_layer.pharmacy_claim`` and ``input_layer.eligibility``
+        — the same tables the CCLF loader writes.  Off by default
+        because it replaces whatever source last wrote them.
 
     Returns
     -------
@@ -89,9 +96,44 @@ def load_bcda(
         stats[table_ref] = count
 
     con.close()
+    if run_pipeline and stats:
+        stats.update(_run_bcda_pipeline(db_path))
     return stats
 
 
+BCDA_PIPELINE_OUTPUTS = {
+    "bcda.medical_claim": "input_layer.medical_claim",
+    "bcda.pharmacy_claim": "input_layer.pharmacy_claim",
+    "bcda.eligibility": "input_layer.eligibility",
+}
+
+
+def _run_bcda_pipeline(database: str) -> dict[str, int]:
+    """Run ``bcda.pipe.cclf`` against *database* and save the input_layer tables.
+
+    Every expression's contract is validated by the runner (column set
+    must match exactly), so a successful run is also the proof that the
+    v3-flattened tables carry what the connector reads.
+    """
+    import narwhals as nw
+
+    from aco.lake.context import DuckDBContext
+    from bcda.pipe.cclf import pipeline
+
+    ctx = DuckDBContext(database=database, read_only=False)
+    cache = pipeline.run(ctx.load)
+    out: dict[str, int] = {}
+    for source, target in BCDA_PIPELINE_OUTPUTS.items():
+        if source in cache:
+            df = cache[source]
+            # narwhalify'd expressions hand back native frames; save wants narwhals.
+            if not isinstance(df, (nw.DataFrame, nw.LazyFrame)):
+                df = nw.from_native(df)
+            ctx.save(target, df, mode="replace")
+            out[target] = len(cache[source])
+    return out
+
+
 def _find_latest_export(store_path: Path) -> list[Path]:
     """Export directories holding the newest file of each resource type.
 

src/aco/pipe/__init__.py

@@ -33,3 +33,12 @@ registry: dict[str, Pipeline] = {
     "quality_measures": quality_measures.pipeline,
     "readmissions": readmissions.pipeline,
 }
+
+# The BCDA connector lives in the optional ``bcda`` extra; register it
+# when installed so ``stack run bcda`` works like ``stack run cclf``.
+try:
+    from bcda.pipe import cclf as _bcda_cclf
+except ImportError:  # pragma: no cover — stack[bcda] not installed
+    _bcda_cclf = None
+if _bcda_cclf is not None:
+    registry["bcda"] = _bcda_cclf.pipeline

src/bcda/express/v3.py

@@ -69,6 +69,7 @@ NDC = "http://hl7.org/fhir/sid/ndc"
 HIPPS = "HIPPSCodes"  # https://www.cms.gov/.../ProspMedicareFeeSvcPmtGen/HIPPSCodes
 HCPCS_SYSTEMS = ("http://www.ama-assn.org/go/cpt", "HCPCSReleaseCodeSets")
 POS = "Place_of_Service_Code_Set"
+NUBC_TOB = "https://www.nubc.org/CodeSystem/TypeOfBill"
 
 # CCW BENE_RACE_CD from the US Core OMB race/ethnicity codes
 # (CCW codebook: 0 Unknown, 1 White, 2 Black, 3 Other, 4 Asian/Pacific
@@ -1070,6 +1071,7 @@ def flatten_eob(resource: dict) -> tuple[ExplanationOfBenefit, list[EobItem]]:
             if v is not None:
                 for col in cols:
                     kwargs[col] = v
+        kwargs["clm_bill_type_cd"] = _coding(tob.get("code"), NUBC_TOB)
     if kwargs.get("bene_ptnt_stus_cd") is None:
         kwargs["bene_ptnt_stus_cd"] = _si_code_by_system(
             infos, CS + "CLM-MDCR-NCH-PTNT-STUS-IND-CD"

src/bcda/pipe/cclf.py

@@ -1,30 +1,32 @@
 """BCDA FHIR R4 connector pipeline — EOB + Patient + Coverage → input_layer.
 
-Transforms flattened BCDA FHIR R4 resources into three consolidated
-output tables:
-  - input_layer.medical_claim  (Part A + Part B carrier + Part B DME)
+Transforms the flattened BCDA tables (v2 or v3 exports) into the three
+input_layer tables on the canonical ``aco.table.input_layer`` contracts:
+  - input_layer.medical_claim  (institutional + professional + DME)
   - input_layer.pharmacy_claim (Part D)
   - input_layer.eligibility    (Patient + Coverage)
 
-Pipeline expressions (12 total)::
+Pipeline expressions (13 total)::
 
-    stg_beneficiary_xref              Patient → MBI identity
-    stg_part_a_header                 EOB(V) + xref
+    stg_beneficiary_xref              Patient → current MBI
+    stg_claim_header                  EOB → final action, claim family,
+                                      cross-source dedupe, MBI
+    stg_part_a_header                 family = institutional
     int_institutional_medical_claim   header + items → medical_claim
-    stg_part_b_physician_header       EOB(O) + xref
+    stg_part_b_physician_header       family = professional
     int_physician_medical_claim       header + items → medical_claim
-    stg_part_b_dme_header             EOB(M) + xref
+    stg_part_b_dme_header             family = dme
     int_dme_medical_claim             header + items → medical_claim
-    medical_claim                     union inst+phys+dme
-    stg_part_d_header                 EOB(P) + xref
+    medical_claim                     union inst + phys + dme
+    stg_part_d_header                 family = pharmacy
     pharmacy_claim                    Part D → pharmacy_claim
-    stg_beneficiary_demographics      Patient + Coverage
+    stg_beneficiary_demographics      Patient + Coverage (Part A row, Dual row)
     eligibility                       demographics → eligibility
 
-Source data — BCDA bulk FHIR export (v2 or v3, flattened to bcda.*):
-  - ExplanationOfBenefit (header + items)
-  - Patient
-  - Coverage
+v3 semantics (routing on EOB-TYPE / CLM-TYPE-CD, Final-Action filter,
+SharedSystem→NationalClaimsHistory dedupe by claim control number,
+consolidated line amounts, admission period, NUBC type of bill, Coverage
+period per plan) are documented in ``bcda.express.cclf``.
 """
 
 from __future__ import annotations
@@ -55,6 +57,7 @@ def _build_pipeline():
         BcdaPharmacyClaim,
         BcdaStgBeneficiaryDemographics,
         BcdaStgBeneficiaryXref,
+        BcdaStgClaimHeader,
         BcdaStgPartAHeader,
         BcdaStgPartBDmeHeader,
         BcdaStgPartBPhysicianHeader,
@@ -76,26 +79,42 @@ def _build_pipeline():
                 after=["bcda.patient"],
                 refs=_REFS,
                 description=(
-                    "Deduplicates BCDA Patient to one row per "
-                    "beneficiary with the current MBI. Unlike the "
-                    "CCLF pipeline, no CCLF9-style crosswalk is "
-                    "needed — BCDA delivers current MBIs directly."
+                    "One row per beneficiary with the current MBI. v2 "
+                    "delivers it in bene_mbi_id; v3's flattener picks the "
+                    "us-mbi identifier without a period end. bene_id is "
+                    "the CCW BENE_ID (v2) or the IDR BENE_SK (v3)."
                 ),
             ),
-            # -- Part A institutional claims --------------------------
+            # -- Claim header: final action, family, dedupe -----------
             Expr(
-                name="bcda._stg_part_a_header",
-                fn=ex.stg_part_a_header,
-                output=BcdaStgPartAHeader,
+                name="bcda._stg_claim_header",
+                fn=ex.stg_claim_header,
+                output=BcdaStgClaimHeader,
                 after=[
                     "bcda.explanation_of_benefit",
                     "bcda._stg_beneficiary_xref",
                 ],
                 refs=_REFS,
                 description=(
-                    "Filters EOB to Part A institutional claims "
-                    "(nch_near_line_rec_ident_cd='V') and resolves "
-                    "beneficiary MBI via the xref."
+                    "Drops NotFinalAction claims (v3 Final-Action tag), "
+                    "routes each EOB to institutional / professional / dme "
+                    "/ pharmacy from the v3 EOB-TYPE and CLM-TYPE-CD "
+                    "codings (v2 near-line RIC as last resort), collapses "
+                    "a claim seen in two sources by claim control number "
+                    "(NationalClaimsHistory over SharedSystem, newest "
+                    "last), and resolves the current MBI."
+                ),
+            ),
+            # -- Institutional claims ---------------------------------
+            Expr(
+                name="bcda._stg_part_a_header",
+                fn=ex.stg_part_a_header,
+                output=BcdaStgPartAHeader,
+                after=["bcda._stg_claim_header"],
+                refs=_REFS,
+                description=(
+                    "Institutional claims: EOB-TYPE INPATIENT, OUTPATIENT, "
+                    "SNF, HHA, HOSPICE (CLM-TYPE-CD 10–64)."
                 ),
             ),
             Expr(
@@ -105,42 +124,35 @@ def _build_pipeline():
                 after=["bcda._stg_part_a_header", "bcda.eob_item"],
                 refs=_REFS,
                 description=(
-                    "Joins Part A header + line items into "
-                    "institutional medical_claim rows with bill type, "
-                    "DRG, revenue center, 12 inline diagnosis codes, "
-                    "and 1 procedure code from the EOB header."
+                    "Institutional header + revenue-centre lines → "
+                    "medical_claim: admission period, NUBC type of bill, "
+                    "MS-DRG, revenue centre, 12 header diagnoses with POA "
+                    "on the first, first ICD procedure, line paid / charge "
+                    "/ coinsurance / deductible from the v3 Adjudication "
+                    "amounts, attending→operating→rendering NPI, billing "
+                    "Organization NPI and tax number."
                 ),
             ),
-            # -- Part B physician claims ------------------------------
+            # -- Part B carrier claims --------------------------------
             Expr(
                 name="bcda._stg_part_b_physician_header",
                 fn=ex.stg_part_b_physician_header,
                 output=BcdaStgPartBPhysicianHeader,
-                after=[
-                    "bcda.explanation_of_benefit",
-                    "bcda._stg_beneficiary_xref",
-                ],
+                after=["bcda._stg_claim_header"],
                 refs=_REFS,
-                description=(
-                    "Filters EOB to Part B carrier/physician claims "
-                    "(nch_near_line_rec_ident_cd='O') and resolves "
-                    "beneficiary MBI."
-                ),
+                description="Part B carrier claims: EOB-TYPE CARRIER (CLM-TYPE-CD 71, 72).",
             ),
             Expr(
                 name="bcda._int_physician_medical_claim",
                 fn=ex.int_physician_medical_claim,
                 output=BcdaIntPhysicianMedicalClaim,
-                after=[
-                    "bcda._stg_part_b_physician_header",
-                    "bcda.eob_item",
-                ],
+                after=["bcda._stg_part_b_physician_header", "bcda.eob_item"],
                 refs=_REFS,
                 description=(
-                    "Joins Part B physician header + line items into "
-                    "professional medical_claim rows with place of "
-                    "service, rendering NPI/TIN, and 12 inline "
-                    "diagnosis codes."
+                    "Carrier header + lines → professional medical_claim: "
+                    "place of service, HCPCS + modifiers, rendering NPI and "
+                    "tax number, billing NPI, line paid / allowed / "
+                    "submitted / coinsurance / deductible."
                 ),
             ),
             # -- Part B DME claims ------------------------------------
@@ -148,16 +160,9 @@ def _build_pipeline():
                 name="bcda._stg_part_b_dme_header",
                 fn=ex.stg_part_b_dme_header,
                 output=BcdaStgPartBDmeHeader,
-                after=[
-                    "bcda.explanation_of_benefit",
-                    "bcda._stg_beneficiary_xref",
-                ],
+                after=["bcda._stg_claim_header"],
                 refs=_REFS,
-                description=(
-                    "Filters EOB to Part B DME claims "
-                    "(nch_near_line_rec_ident_cd='M') and resolves "
-                    "beneficiary MBI."
-                ),
+                description="Part B DME claims: EOB-TYPE DME (CLM-TYPE-CD 81, 82).",
             ),
             Expr(
                 name="bcda._int_dme_medical_claim",
@@ -166,12 +171,12 @@ def _build_pipeline():
                 after=["bcda._stg_part_b_dme_header", "bcda.eob_item"],
                 refs=_REFS,
                 description=(
-                    "Joins Part B DME header + line items into "
-                    "professional medical_claim rows. DME claims "
-                    "carry no diagnosis or procedure codes."
+                    "DME header + lines → professional medical_claim; "
+                    "rendering NPI falls back to the ordering provider, "
+                    "billing NPI to the pay-to provider."
                 ),
             ),
-            # -- Final medical_claim (union) --------------------------
+            # -- Union --------------------------------------------------
             Expr(
                 name="bcda.medical_claim",
                 fn=ex.medical_claim,
@@ -183,25 +188,18 @@ def _build_pipeline():
                 ],
                 refs=_REFS,
                 description=(
-                    "Unions institutional + professional + DME into "
-                    "the final medical_claim table."
+                    "Union of institutional, carrier and DME lines on the "
+                    "input_layer.medical_claim contract."
                 ),
             ),
-            # -- Part D pharmacy claims -------------------------------
+            # -- Part D ----------------------------------------------
             Expr(
                 name="bcda._stg_part_d_header",
                 fn=ex.stg_part_d_header,
                 output=BcdaStgPartDHeader,
-                after=[
-                    "bcda.explanation_of_benefit",
-                    "bcda._stg_beneficiary_xref",
-                ],
+                after=["bcda._stg_claim_header"],
                 refs=_REFS,
-                description=(
-                    "Filters EOB to Part D pharmacy claims "
-                    "(nch_near_line_rec_ident_cd='P') and resolves "
-                    "beneficiary MBI."
-                ),
+                description="Part D claims: EOB-TYPE PDE (CLM-TYPE-CD 1–4).",
             ),
             Expr(
                 name="bcda.pharmacy_claim",
@@ -210,12 +208,14 @@ def _build_pipeline():
                 after=["bcda._stg_part_d_header", "bcda.eob_item"],
                 refs=_REFS,
                 description=(
-                    "Joins Part D header + line items into "
-                    "pharmacy_claim with NDC, quantity, days supply, "
-                    "prescribing and dispensing provider NPIs."
+                    "Part D header + line → pharmacy_claim: PDE id, "
+                    "prescribing care-team NPI, dispensing pharmacy NPI "
+                    "(contained billing Organization), NDC and NDC "
+                    "quantity, days supply, fill number, total drug cost "
+                    "as charge, paid-by-patient as copayment."
                 ),
             ),
-            # -- Eligibility ------------------------------------------
+            # -- Eligibility -------------------------------------------
             Expr(
                 name="bcda._stg_beneficiary_demographics",
                 fn=ex.stg_beneficiary_demographics,
@@ -223,9 +223,10 @@ def _build_pipeline():
                 after=["bcda.patient", "bcda.coverage"],
                 refs=_REFS,
                 description=(
-                    "Joins Patient + Coverage for demographics and "
-                    "enrollment data. Deduplicates to one row per "
-                    "beneficiary MBI."
+                    "Patient joined to its Part A Coverage row (Part B "
+                    "fallback) for Medicare status, entitlement reason, "
+                    "enrollment period and status, plus the v3 Dual "
+                    "Medicare/Medicaid row for dual status."
                 ),
             ),
             Expr(
@@ -235,8 +236,8 @@ def _build_pipeline():
                 after=["bcda._stg_beneficiary_demographics"],
                 refs=_REFS,
                 description=(
-                    "Maps Patient + Coverage to the eligibility "
-                    "schema with gender and race crosswalks."
+                    "Demographics → input_layer.eligibility with gender, "
+                    "race and enrollment-status crosswalks."
                 ),
             ),
         ]

src/bcda/table/cclf_pipe.py

@@ -1,383 +1,175 @@
-"""SQLTable contracts for the BCDA FHIR R4 connector pipeline intermediates.
+"""SQLTable contracts for the BCDA FHIR R4 connector pipeline.
 
-Each class corresponds to one ``Expr`` in ``bcda.pipe.cclf`` and documents
-the output schema produced by the matching ``bcda.express.cclf`` function.
+Each class is the output contract of one ``Expr`` in ``bcda.pipe.cclf``
+and is checked column-for-column by ``aco.pipe.runner`` against what
+the matching ``bcda.express.cclf`` function returns.
+
+The contracts inherit rather than restate:
+
+- the claim-header stages pass every ``bcda.explanation_of_benefit``
+  column through plus ``claim_family`` and the current MBI, so they
+  subclass :class:`bcda.table.eob.ExplanationOfBenefit`;
+- the three outputs are the canonical Tuva-shaped input_layer tables,
+  so they subclass ``aco.table.input_layer.InputLayer*`` — the same
+  schema the CCLF pipeline lands, which is what lets the downstream
+  pipelines run unchanged on either source.
 """
 
 from __future__ import annotations
 
-from datetime import date, datetime
+from datetime import date
 
+from aco.table.input_layer import (
+    InputLayerEligibility,
+    InputLayerMedicalClaim,
+    InputLayerPharmacyClaim,
+)
+from bcda.table.eob import ExplanationOfBenefit
 from conf.table_base import SQLTable
 
 # ── Stage 1: Beneficiary XREF (Patient) ──────────────────────────────────────
 
 
 class BcdaStgBeneficiaryXref(SQLTable):
-    """bcda._stg_beneficiary_xref — deduplicated MBI identity from BCDA Patient."""
+    """bcda._stg_beneficiary_xref — one row per beneficiary with the current MBI."""
 
     __schema__ = "bcda"
     __tablename__ = "_stg_beneficiary_xref"
 
     bene_id: str | None = None
-    """Internal BCDA beneficiary identifier."""
+    """Beneficiary key: CCW BENE_ID (v2) or IDR BENE_SK (v3 Patient.id)."""
 
     current_bene_mbi_id: str | None = None
-    """Current Medicare Beneficiary Identifier (delivered directly by BCDA)."""
+    """Current Medicare Beneficiary Identifier."""
 
 
-# ── Stage 2: Part A institutional claim header ────────────────────────────────
+# ── Stage 2: Claim header — final action, family, dedupe, MBI ────────────────
 
 
-class BcdaStgPartAHeader(SQLTable):
-    """bcda._stg_part_a_header — EOB Part A filtered and MBI resolved."""
+class BcdaStgClaimHeader(ExplanationOfBenefit):
+    """bcda._stg_claim_header — final-action EOB headers tagged with a family.
+
+    Every ``bcda.explanation_of_benefit`` column plus the two below.
+    """
+
+    __schema__ = "bcda"
+    __tablename__ = "_stg_claim_header"
+
+    claim_family: str | None = None
+    """institutional · professional · dme · pharmacy (see
+    ``bcda.express.cclf._claim_family``)."""
+
+    current_bene_mbi_id: str | None = None
+    """Current MBI resolved through ``_stg_beneficiary_xref``."""
+
+
+class BcdaStgPartAHeader(BcdaStgClaimHeader):
+    """bcda._stg_part_a_header — institutional claims."""
 
     __schema__ = "bcda"
     __tablename__ = "_stg_part_a_header"
 
-    clm_id: str | None = None
-    current_bene_mbi_id: str | None = None
-    bene_id: str | None = None
-    nch_near_line_rec_ident_cd: str | None = None
-    clm_from_dt: date | None = None
-    clm_thru_dt: date | None = None
-    clm_admsn_dt: date | None = None
-    nch_bene_dschrg_dt: date | None = None
-    clm_admsn_src_cd: str | None = None
-    clm_admsn_type_cd: str | None = None
-    ptnt_dschrg_stus_cd: str | None = None
-    clm_bill_fac_type_cd: str | None = None
-    clm_bill_clsfctn_cd: str | None = None
-    clm_bill_freq_cd: str | None = None
-    clm_drg_cd: str | None = None
-    dgns_prcdr_icd_ind: str | None = None
-    clm_pmt_amt: float | None = None
-    clm_tot_chrg_amt: float | None = None
-    org_npi_num: str | None = None
-    at_physn_npi: str | None = None
-    op_physn_npi: str | None = None
 
-
-# ── Stage 3: Institutional medical_claim (Part A header + items) ──────────────
-
-
-class BcdaIntInstitutionalMedicalClaim(SQLTable):
-    """bcda._int_institutional_medical_claim — Part A joined to medical_claim schema."""
-
-    __schema__ = "bcda"
-    __tablename__ = "_int_institutional_medical_claim"
-
-    claim_id: str | None = None
-    claim_line_number: int | None = None
-    claim_type: str | None = None
-    person_id: str | None = None
-    member_id: str | None = None
-    payer: str | None = None
-    plan: str | None = None
-    claim_start_date: date | None = None
-    claim_end_date: date | None = None
-    admission_date: date | None = None
-    discharge_date: date | None = None
-    admit_source_code: str | None = None
-    admit_type_code: str | None = None
-    discharge_disposition_code: str | None = None
-    bill_type_code: str | None = None
-    drg_code_type: str | None = None
-    drg_code: str | None = None
-    revenue_center_code: str | None = None
-    hcpcs_code: str | None = None
-    rendering_npi: str | None = None
-    billing_npi: str | None = None
-    facility_npi: str | None = None
-    paid_amount: float | None = None
-    allowed_amount: float | None = None
-    charge_amount: float | None = None
-    diagnosis_code_type: str | None = None
-    diagnosis_code_1: str | None = None
-    diagnosis_code_2: str | None = None
-    diagnosis_code_3: str | None = None
-    procedure_code_type: str | None = None
-    procedure_code_1: str | None = None
-    in_network_flag: int | None = None
-    data_source: str | None = None
-    file_name: str | None = None
-    file_date: date | None = None
-    ingest_datetime: datetime | None = None
-
-
-# ── Stage 4: Part B physician claim header ────────────────────────────────────
-
-
-class BcdaStgPartBPhysicianHeader(SQLTable):
-    """bcda._stg_part_b_physician_header — EOB Part B carrier filtered and MBI resolved."""
+class BcdaStgPartBPhysicianHeader(BcdaStgClaimHeader):
+    """bcda._stg_part_b_physician_header — Part B carrier claims."""
 
     __schema__ = "bcda"
     __tablename__ = "_stg_part_b_physician_header"
 
-    clm_id: str | None = None
-    current_bene_mbi_id: str | None = None
-    bene_id: str | None = None
-    nch_near_line_rec_ident_cd: str | None = None
-    clm_from_dt: date | None = None
-    clm_thru_dt: date | None = None
-    clm_pmt_amt: float | None = None
-    clm_tot_chrg_amt: float | None = None
-    rndrng_physn_npi: str | None = None
-    org_npi_num: str | None = None
-    dgns_prcdr_icd_ind: str | None = None
 
-
-# ── Stage 5: Professional medical_claim (Part B physician) ───────────────────
-
-
-class BcdaIntPhysicianMedicalClaim(SQLTable):
-    """bcda._int_physician_medical_claim — Part B physician joined to medical_claim schema."""
-
-    __schema__ = "bcda"
-    __tablename__ = "_int_physician_medical_claim"
-
-    claim_id: str | None = None
-    claim_line_number: int | None = None
-    claim_type: str | None = None
-    person_id: str | None = None
-    member_id: str | None = None
-    payer: str | None = None
-    plan: str | None = None
-    claim_start_date: date | None = None
-    claim_end_date: date | None = None
-    claim_line_start_date: date | None = None
-    claim_line_end_date: date | None = None
-    place_of_service_code: str | None = None
-    hcpcs_code: str | None = None
-    hcpcs_modifier_1: str | None = None
-    rendering_npi: str | None = None
-    billing_npi: str | None = None
-    paid_amount: float | None = None
-    allowed_amount: float | None = None
-    charge_amount: float | None = None
-    diagnosis_code_type: str | None = None
-    diagnosis_code_1: str | None = None
-    diagnosis_code_2: str | None = None
-    diagnosis_code_3: str | None = None
-    in_network_flag: int | None = None
-    data_source: str | None = None
-    file_name: str | None = None
-    file_date: date | None = None
-    ingest_datetime: datetime | None = None
-
-
-# ── Stage 6: Part B DME claim header ─────────────────────────────────────────
-
-
-class BcdaStgPartBDmeHeader(SQLTable):
-    """bcda._stg_part_b_dme_header — EOB Part B DME filtered and MBI resolved."""
+class BcdaStgPartBDmeHeader(BcdaStgClaimHeader):
+    """bcda._stg_part_b_dme_header — Part B DME claims."""
 
     __schema__ = "bcda"
     __tablename__ = "_stg_part_b_dme_header"
 
-    clm_id: str | None = None
-    current_bene_mbi_id: str | None = None
-    bene_id: str | None = None
-    nch_near_line_rec_ident_cd: str | None = None
-    clm_from_dt: date | None = None
-    clm_thru_dt: date | None = None
-    clm_pmt_amt: float | None = None
-    clm_tot_chrg_amt: float | None = None
-    rndrng_physn_npi: str | None = None
-    org_npi_num: str | None = None
 
-
-# ── Stage 7: DME medical_claim (Part B DME) ───────────────────────────────────
-
-
-class BcdaIntDmeMedicalClaim(SQLTable):
-    """bcda._int_dme_medical_claim — Part B DME joined to medical_claim schema."""
-
-    __schema__ = "bcda"
-    __tablename__ = "_int_dme_medical_claim"
-
-    claim_id: str | None = None
-    claim_line_number: int | None = None
-    claim_type: str | None = None
-    person_id: str | None = None
-    member_id: str | None = None
-    payer: str | None = None
-    plan: str | None = None
-    claim_start_date: date | None = None
-    claim_end_date: date | None = None
-    claim_line_start_date: date | None = None
-    claim_line_end_date: date | None = None
-    hcpcs_code: str | None = None
-    rendering_npi: str | None = None
-    billing_npi: str | None = None
-    paid_amount: float | None = None
-    allowed_amount: float | None = None
-    charge_amount: float | None = None
-    in_network_flag: int | None = None
-    data_source: str | None = None
-    file_name: str | None = None
-    file_date: date | None = None
-    ingest_datetime: datetime | None = None
-
-
-# ── Stage 8: Final medical_claim (union inst + phys + dme) ───────────────────
-
-
-class BcdaMedicalClaim(SQLTable):
-    """bcda.medical_claim — unioned Part A + Part B carrier + Part B DME claims."""
-
-    __schema__ = "bcda"
-    __tablename__ = "medical_claim"
-
-    claim_id: str | None = None
-    claim_line_number: int | None = None
-    claim_type: str | None = None
-    person_id: str | None = None
-    member_id: str | None = None
-    payer: str | None = None
-    plan: str | None = None
-    claim_start_date: date | None = None
-    claim_end_date: date | None = None
-    admission_date: date | None = None
-    discharge_date: date | None = None
-    admit_source_code: str | None = None
-    admit_type_code: str | None = None
-    discharge_disposition_code: str | None = None
-    place_of_service_code: str | None = None
-    bill_type_code: str | None = None
-    drg_code_type: str | None = None
-    drg_code: str | None = None
-    revenue_center_code: str | None = None
-    hcpcs_code: str | None = None
-    rendering_npi: str | None = None
-    billing_npi: str | None = None
-    facility_npi: str | None = None
-    paid_amount: float | None = None
-    allowed_amount: float | None = None
-    charge_amount: float | None = None
-    diagnosis_code_type: str | None = None
-    diagnosis_code_1: str | None = None
-    diagnosis_code_2: str | None = None
-    diagnosis_code_3: str | None = None
-    procedure_code_type: str | None = None
-    procedure_code_1: str | None = None
-    in_network_flag: int | None = None
-    data_source: str | None = None
-    file_name: str | None = None
-    file_date: date | None = None
-    ingest_datetime: datetime | None = None
-
-
-# ── Stage 9: Part D pharmacy claim header ────────────────────────────────────
-
-
-class BcdaStgPartDHeader(SQLTable):
-    """bcda._stg_part_d_header — EOB Part D filtered and MBI resolved."""
+class BcdaStgPartDHeader(BcdaStgClaimHeader):
+    """bcda._stg_part_d_header — Part D pharmacy claims."""
 
     __schema__ = "bcda"
     __tablename__ = "_stg_part_d_header"
 
-    clm_id: str | None = None
-    current_bene_mbi_id: str | None = None
-    bene_id: str | None = None
-    nch_near_line_rec_ident_cd: str | None = None
-    clm_from_dt: date | None = None
-    clm_thru_dt: date | None = None
-    clm_pmt_amt: float | None = None
-    clm_tot_chrg_amt: float | None = None
-    prscrbr_id: str | None = None
-    rx_srvc_rfrnc_num: str | None = None
+
+# ── Stage 3: medical_claim (input_layer shape) ───────────────────────────────
 
 
-# ── Stage 10: Final pharmacy_claim (Part D) ───────────────────────────────────
+class BcdaIntInstitutionalMedicalClaim(InputLayerMedicalClaim):
+    """bcda._int_institutional_medical_claim — institutional lines."""
+
+    __schema__ = "bcda"
+    __tablename__ = "_int_institutional_medical_claim"
 
 
-class BcdaPharmacyClaim(SQLTable):
-    """bcda.pharmacy_claim — Part D joined to pharmacy_claim schema."""
+class BcdaIntPhysicianMedicalClaim(InputLayerMedicalClaim):
+    """bcda._int_physician_medical_claim — carrier lines."""
+
+    __schema__ = "bcda"
+    __tablename__ = "_int_physician_medical_claim"
+
+
+class BcdaIntDmeMedicalClaim(InputLayerMedicalClaim):
+    """bcda._int_dme_medical_claim — DME lines."""
+
+    __schema__ = "bcda"
+    __tablename__ = "_int_dme_medical_claim"
+
+
+class BcdaMedicalClaim(InputLayerMedicalClaim):
+    """bcda.medical_claim — union of the three families; saved to input_layer.medical_claim."""
+
+    __schema__ = "bcda"
+    __tablename__ = "medical_claim"
+
+
+# ── Stage 4: pharmacy_claim ──────────────────────────────────────────────────
+
+
+class BcdaPharmacyClaim(InputLayerPharmacyClaim):
+    """bcda.pharmacy_claim — Part D; saved to input_layer.pharmacy_claim."""
 
     __schema__ = "bcda"
     __tablename__ = "pharmacy_claim"
 
-    claim_id: str | None = None
-    claim_line_number: int | None = None
-    person_id: str | None = None
-    member_id: str | None = None
-    payer: str | None = None
-    plan: str | None = None
-    prescribing_provider_npi: str | None = None
-    dispensing_provider_npi: str | None = None
-    dispensing_date: date | None = None
-    ndc_code: str | None = None
-    quantity: float | None = None
-    days_supply: int | None = None
-    refills: int | None = None
-    paid_date: date | None = None
-    paid_amount: float | None = None
-    allowed_amount: float | None = None
-    coinsurance_amount: float | None = None
-    copayment_amount: float | None = None
-    deductible_amount: float | None = None
-    in_network_flag: int | None = None
-    data_source: str | None = None
-    file_name: str | None = None
-    file_date: date | None = None
-    ingest_datetime: datetime | None = None
 
-
-# ── Stage 11: Beneficiary demographics staging (Patient + Coverage) ───────────
+# ── Stage 5: eligibility ─────────────────────────────────────────────────────
 
 
 class BcdaStgBeneficiaryDemographics(SQLTable):
-    """bcda._stg_beneficiary_demographics — Patient + Coverage joined and deduplicated."""
+    """bcda._stg_beneficiary_demographics — Patient plus its enrollment Coverage row."""
 
     __schema__ = "bcda"
     __tablename__ = "_stg_beneficiary_demographics"
 
-    current_bene_mbi_id: str | None = None
     bene_id: str | None = None
-    bene_birth_dt: date | None = None
+    bene_mbi_id: str | None = None
     bene_sex_cd: str | None = None
     bene_race_cd: str | None = None
-    bene_fips_state_cd: str | None = None
-    bene_county_cd: str | None = None
-    bene_zip_cd: str | None = None
+    bene_dob: date | None = None
+    bene_death_dt: date | None = None
     bene_1st_name: str | None = None
+    bene_midl_name: str | None = None
     bene_last_name: str | None = None
-    crnt_bene_mdcr_stus_cd: str | None = None
-    covr_yr_mo: str | None = None
+    bene_line_1_adr: str | None = None
+    geo_zip_plc_name: str | None = None
+    geo_usps_state_cd: str | None = None
+    geo_zip5_cd: str | None = None
+    coverage_plan: str | None = None
+    """Which Coverage row supplied the enrollment fields (Part A, else Part B)."""
+    coverage_status: str | None = None
+    """Coverage.status of that row (active / cancelled)."""
+    ms_cd: str | None = None
+    orec: str | None = None
+    dual_01: str | None = None
+    """v2 January dual status (NULL on v3 rows)."""
+    enrollment_start_date: date | None = None
+    enrollment_end_date: date | None = None
+    dual_stus_cd_dual: str | None = None
+    """v3 dual status from the Dual Medicare/Medicaid Coverage row."""
 
 
-# ── Stage 12: Final eligibility (Patient + Coverage) ─────────────────────────
-
-
-class BcdaEligibility(SQLTable):
-    """bcda.eligibility — Patient + Coverage mapped to eligibility schema."""
+class BcdaEligibility(InputLayerEligibility):
+    """bcda.eligibility — saved to input_layer.eligibility."""
 
     __schema__ = "bcda"
     __tablename__ = "eligibility"
-
-    person_id: str | None = None
-    member_id: str | None = None
-    gender: str | None = None
-    race: str | None = None
-    birth_date: date | None = None
-    death_date: date | None = None
-    death_flag: int | None = None
-    enrollment_start_date: date | None = None
-    enrollment_end_date: date | None = None
-    payer: str | None = None
-    plan: str | None = None
-    original_reason_entitlement_code: str | None = None
-    dual_status_code: str | None = None
-    medicare_status_code: str | None = None
-    first_name: str | None = None
-    last_name: str | None = None
-    address: str | None = None
-    city: str | None = None
-    state: str | None = None
-    zip_code: str | None = None
-    phone: str | None = None
-    data_source: str | None = None
-    file_name: str | None = None
-    file_date: date | None = None
-    ingest_datetime: datetime | None = None

src/bcda/table/eob.py

@@ -414,6 +414,12 @@ class ExplanationOfBenefit(SQLTable):
     clm_sbmt_frmt_cd: str | None = None
     """Submission format (supportingInfo CLM_SBMT_FRMT_CD)."""
 
+    clm_bill_type_cd: str | None = None
+    """NUBC type of bill, 4 characters (supportingInfo typeofbill coding
+    https://www.nubc.org/CodeSystem/TypeOfBill) — the v3 CLM-BILL-FREQ-CD
+    coding carries the last three characters, so this is the only place
+    the whole code appears."""
+
     clm_blg_prvdr_name: str | None = None
     """Billing provider name (contained Organization.name)."""
 

src/cli/load.py

@@ -52,6 +52,11 @@ def bcda(
         "--skip-flatten",
         help="Skip flattening, load existing Parquet.",
     ),
+    run_pipeline: bool = typer.Option(
+        False,
+        "--run-pipeline",
+        help="Also run the bcda connector pipeline into input_layer.*",
+    ),
 ) -> None:
     """Flatten BCDA FHIR exports and load into DuckDB.
 
@@ -68,6 +73,7 @@ def bcda(
             ndjson_dir=ndjson_dir,
             database=database,
             skip_flatten=skip_flatten,
+            run_pipeline=run_pipeline,
         )
     except FileNotFoundError as e:
         typer.echo(f"Error: {e}", err=True)

tests/aco/test_load_bcda.py

@@ -95,12 +95,17 @@ class TestLoadV3EndToEnd:
             return real_path(key)
 
         with patch("conf.path", side_effect=fake_path):
-            stats = load_bcda(database=str(db))
+            stats = load_bcda(database=str(db), run_pipeline=True)
         assert stats == {
             "bcda.patient": 94,
             "bcda.explanation_of_benefit": 538,
             "bcda.eob_item": 2464,
             "bcda.coverage": 175,
+            # the connector pipeline ran and landed input_layer.*; the
+            # duplicated EOB part collapses on claim control number
+            "input_layer.medical_claim": 1021,
+            "input_layer.pharmacy_claim": 102,
+            "input_layer.eligibility": 94,
         }
         con = duckdb.connect(str(db), read_only=True)
         plans = dict(
@@ -121,4 +126,17 @@ class TestLoadV3EndToEnd:
             ).fetchone()[0]
             == 0
         )
+        assert (
+            con.execute(
+                "select count(distinct claim_id) from input_layer.medical_claim"
+            ).fetchone()[0]
+            == 131
+        )
+        assert (
+            con.execute(
+                "select count(*) from input_layer.medical_claim "
+                "where claim_type = 'institutional' and length(bill_type_code) <> 4"
+            ).fetchone()[0]
+            == 0
+        )
         con.close()

tests/bcda/test_cclf_v3.py

@@ -0,0 +1,923 @@
+"""bcda.express.cclf — the connector pipeline on v2- and v3-shaped tables.
+
+Routing, final-action filtering, cross-source dedupe, the shared
+medical_claim mapping, Part D, demographics and eligibility.  Fixtures
+are minimal frames; the official-sample run lives in test_pipe_cclf.
+"""
+
+from __future__ import annotations
+
+from datetime import date
+
+import narwhals as nw
+import polars as pl
+import pytest
+
+from bcda.express import cclf as ex
+from bcda.express.cclf import (
+    eligibility,
+    int_dme_medical_claim,
+    int_institutional_medical_claim,
+    int_physician_medical_claim,
+    medical_claim,
+    pharmacy_claim,
+    stg_beneficiary_demographics,
+    stg_beneficiary_xref,
+    stg_claim_header,
+    stg_part_a_header,
+    stg_part_b_dme_header,
+    stg_part_b_physician_header,
+    stg_part_d_header,
+)
+from bcda.table.cclf_pipe import (
+    BcdaEligibility,
+    BcdaMedicalClaim,
+    BcdaPharmacyClaim,
+    BcdaStgBeneficiaryDemographics,
+    BcdaStgClaimHeader,
+)
+
+# ── fixtures ─────────────────────────────────────────────────
+
+
+@pytest.fixture
+def xref() -> pl.DataFrame:
+    return stg_beneficiary_xref(
+        bcda__patient=pl.DataFrame(
+            {"bene_id": ["B1", "B2"], "bene_mbi_id": ["MBI-1", "MBI-2"]}
+        )
+    )
+
+
+def _eob(rows: list[dict]) -> pl.DataFrame:
+    """Header frame from sparse dicts; missing columns are filled by the stage."""
+    cols: dict[str, list] = {}
+    for r in rows:
+        for k in r:
+            cols.setdefault(k, [])
+    for r in rows:
+        for k in cols:
+            cols[k].append(r.get(k))
+    return pl.DataFrame(cols)
+
+
+@pytest.fixture
+def v3_headers() -> pl.DataFrame:
+    """Seven v3 claims: one per family, a NotFinalAction, and a SharedSystem/NCH pair."""
+    return _eob(
+        [
+            {
+                "clm_id": "I1",
+                "clm_cntl_num": "CCN-I1",
+                "bene_id": "B1",
+                "eob_type": "INPATIENT",
+                "clm_type_cd": "60",
+                "system_type": "NationalClaimsHistory",
+                "final_action": "FinalAction",
+                "last_updated": "2026-01-02",
+                "clm_from_dt": date(2025, 1, 1),
+                "clm_thru_dt": date(2025, 1, 9),
+                "clm_actv_care_from_dt": date(2024, 12, 31),
+                "clm_dschrg_dt": date(2025, 1, 9),
+                "clm_admsn_src_cd": "7",
+                "clm_admsn_type_cd": "1",
+                "bene_ptnt_stus_cd": "01",
+                "clm_bill_type_cd": "0111",
+                "clm_bill_fac_type_cd": "1",
+                "clm_bill_clsfctn_cd": "11",
+                "clm_bill_freq_cd": "111",
+                "dgns_drg_cd": "470",
+                "dgns_prcdr_icd_ind": "0",
+                "clm_dgns_1_cd": "I10",
+                "clm_poa_ind": "Y",
+                "clm_prcdr_cd": "0SRC0J9",
+                "clm_prcdr_prfrm_dt": date(2025, 1, 2),
+                "atndg_prvdr_npi_num": "ATT",
+                "oprtg_prvdr_npi_num": "OPR",
+                "fac_prvdr_npi_num": "FAC",
+                "clm_blg_prvdr_tax_num": "TIN-FAC",
+                "clm_nch_wkly_proc_dt": date(2025, 1, 20),
+                "clm_mdcr_instnl_tot_chrg_amt": 9999.0,
+            },
+            {
+                "clm_id": "C1",
+                "clm_cntl_num": "CCN-C1",
+                "bene_id": "B1",
+                "eob_type": "CARRIER",
+                "clm_type_cd": "71",
+                "system_type": "NationalClaimsHistory",
+                "final_action": "FinalAction",
+                "last_updated": "2026-01-02",
+                "clm_from_dt": date(2025, 2, 1),
+                "clm_thru_dt": date(2025, 2, 1),
+                "dgns_prcdr_icd_ind": "0",
+                "clm_dgns_1_cd": "E11.9",
+                "rndrg_prvdr_npi_num": "RND",
+                "clm_rndrg_prvdr_tax_num": "TIN-RND",
+                "carr_clm_blg_npi_num": "BLG",
+                "clm_cms_proc_dt": date(2025, 2, 10),
+            },
+            {
+                "clm_id": "D1",
+                "clm_cntl_num": "CCN-D1",
+                "bene_id": "B2",
+                "eob_type": "DME",
+                "clm_type_cd": "82",
+                "system_type": "NationalClaimsHistory",
+                "final_action": "FinalAction",
+                "last_updated": "2026-01-02",
+                "clm_from_dt": date(2025, 3, 1),
+                "clm_thru_dt": date(2025, 3, 1),
+                "ordrg_prvdr_npi_num": "ORD",
+                "payto_prvdr_npi_num": "PAY",
+            },
+            {
+                "clm_id": "P1",
+                "pde_id": "P1",
+                "clm_cntl_num": "CCN-P1",
+                "bene_id": "B2",
+                "eob_type": "PDE",
+                "clm_type_cd": "1",
+                "system_type": "DDPS",
+                "final_action": "FinalAction",
+                "last_updated": "2026-01-02",
+                "clm_from_dt": date(2025, 4, 1),
+                "clm_thru_dt": date(2025, 4, 1),
+                "clm_prsbng_prvdr_gnrc_id_num": "PRS",
+                "fac_prvdr_npi_num": "PHARM",
+                "clm_srvc_prvdr_gnrc_id_num": "LIC",
+                "clm_line_days_suply_qty": 30.0,
+                "clm_line_rx_fill_num": 2.0,
+                "clm_pd_dt": date(2025, 4, 3),
+                "clm_bene_pmt_amt": 7.5,
+                "tot_rx_cst_amt": 120.0,
+            },
+            {
+                "clm_id": "X1",
+                "clm_cntl_num": "CCN-X1",
+                "bene_id": "B1",
+                "eob_type": "SNF",
+                "clm_type_cd": "20",
+                "system_type": "NationalClaimsHistory",
+                "final_action": "NotFinalAction",
+                "last_updated": "2026-01-02",
+                "clm_from_dt": date(2025, 5, 1),
+                "clm_thru_dt": date(2025, 5, 2),
+            },
+            {
+                "clm_id": "S1",
+                "clm_cntl_num": "CCN-PAIR",
+                "bene_id": "B1",
+                "eob_type": "OUTPATIENT",
+                "clm_type_cd": "40",
+                "system_type": "SharedSystem",
+                "final_action": "FinalAction",
+                "last_updated": "2026-01-05",
+                "clm_from_dt": date(2025, 6, 1),
+                "clm_thru_dt": date(2025, 6, 1),
+            },
+            {
+                "clm_id": "N1",
+                "clm_cntl_num": "CCN-PAIR",
+                "bene_id": "B1",
+                "eob_type": "OUTPATIENT",
+                "clm_type_cd": "40",
+                "system_type": "NationalClaimsHistory",
+                "final_action": "FinalAction",
+                "last_updated": "2026-01-01",
+                "clm_from_dt": date(2025, 6, 1),
+                "clm_thru_dt": date(2025, 6, 1),
+            },
+        ]
+    )
+
+
+@pytest.fixture
+def v3_items() -> pl.DataFrame:
+    return _eob(
+        [
+            {
+                "clm_id": "I1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2025, 1, 1),
+                "clm_line_thru_dt": date(2025, 1, 9),
+                "clm_line_prod_rev_ctr_cd": "0120",
+                "clm_line_hcpcs_cd": "99223",
+                "clm_line_srvc_unit_qty": 1.4,
+                "clm_line_cvrd_pd_amt": 100.0,
+                "rev_cntr_tot_chrg_amt": 250.0,
+                "rev_cntr_coinsrnc_wge_adjstd_c": 10.0,
+                "rev_cntr_cash_ddctbl_amt": 5.0,
+            },
+            {
+                "clm_id": "I1",
+                "clm_line_num": 2.0,
+                "clm_line_from_dt": date(2025, 1, 2),
+                "clm_line_thru_dt": date(2025, 1, 2),
+                "clm_line_prod_rev_ctr_cd": "0450",
+                "clm_line_hcpcs_cd": None,
+                "clm_line_srvc_unit_qty": 2.0,
+                "clm_line_cvrd_pd_amt": 50.0,
+                "rev_cntr_tot_chrg_amt": 75.0,
+            },
+            {
+                "clm_id": "C1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2025, 2, 1),
+                "clm_line_thru_dt": date(2025, 2, 1),
+                "clm_line_hcpcs_cd": "99213",
+                "hcpcs_1_mdfr_cd": "25",
+                "clm_pos_cd": "11",
+                "clm_line_srvc_unit_qty": 1.0,
+                "clm_line_cvrd_pd_amt": 60.0,
+                "clm_line_alowd_chrg_amt": 75.0,
+                "clm_line_sbmtd_chrg_amt": 150.0,
+                "clm_line_coinsrnc_amt": 15.0,
+                "clm_line_bene_ptb_ddctbl_amt": 0.0,
+            },
+            {
+                "clm_id": "D1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2025, 3, 1),
+                "clm_line_thru_dt": date(2025, 3, 31),
+                "clm_line_hcpcs_cd": "E1390",
+                "clm_pos_cd": "12",
+                "clm_line_srvc_unit_qty": 1.0,
+                "clm_line_cvrd_pd_amt": 80.0,
+                "clm_line_alowd_chrg_amt": 100.0,
+                "clm_line_sbmtd_chrg_amt": 120.0,
+            },
+            {
+                "clm_id": "P1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2025, 4, 1),
+                "clm_line_thru_dt": date(2025, 4, 1),
+                "clm_line_ndc_cd": "00002143380",
+                "clm_line_ndc_qty": 30.0,
+                "clm_line_srvc_unit_qty": 1.0,
+                "tot_rx_cst_amt": 118.0,
+            },
+            {
+                "clm_id": "N1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2025, 6, 1),
+                "clm_line_thru_dt": date(2025, 6, 1),
+                "clm_line_hcpcs_cd": "G0463",
+                "clm_line_cvrd_pd_amt": 30.0,
+            },
+            {
+                "clm_id": "X1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2025, 5, 1),
+                "clm_line_thru_dt": date(2025, 5, 1),
+                "clm_line_cvrd_pd_amt": 1.0,
+            },
+        ]
+    )
+
+
+@pytest.fixture
+def v2_headers() -> pl.DataFrame:
+    """v2-shaped headers: routed by RIC code, no tags, no control numbers."""
+    return _eob(
+        [
+            {
+                "clm_id": "V1",
+                "bene_id": "B1",
+                "nch_near_line_rec_ident_cd": "V",
+                "clm_from_dt": date(2023, 1, 1),
+                "clm_thru_dt": date(2023, 1, 5),
+                "clm_bill_fac_type_cd": "1",
+                "clm_bill_clsfctn_cd": "1",
+                "clm_bill_freq_cd": "1",
+                "dgns_drg_cd": "MS-DRG470",
+                "oprtg_prvdr_npi_num": "OP",
+                "fac_prvdr_npi_num": "FACV",
+                "clm_mdcr_instnl_tot_chrg_amt": 5000.0,
+            },
+            {
+                "clm_id": "O1",
+                "bene_id": "B1",
+                "nch_near_line_rec_ident_cd": "O",
+                "clm_from_dt": date(2023, 2, 1),
+                "clm_thru_dt": date(2023, 2, 1),
+                "rndrg_prvdr_npi_num": "RN",
+                "carr_clm_blg_npi_num": "BL",
+            },
+            {
+                "clm_id": "M1",
+                "bene_id": "B2",
+                "nch_near_line_rec_ident_cd": "M",
+                "clm_from_dt": date(2023, 3, 1),
+                "clm_thru_dt": date(2023, 3, 1),
+                "ordrg_prvdr_npi_num": "ORD",
+                "payto_prvdr_npi_num": "PAY",
+            },
+            {
+                "clm_id": "P2",
+                "pde_id": "PDE-2",
+                "bene_id": "B2",
+                "nch_near_line_rec_ident_cd": "P",
+                "clm_from_dt": date(2023, 4, 1),
+                "clm_thru_dt": date(2023, 4, 1),
+                "clm_prsbng_prvdr_gnrc_id_num": "PRS2",
+                "clm_srvc_prvdr_gnrc_id_num": "DSP2",
+                "clm_line_days_suply_qty": 30.0,
+                "clm_line_rx_fill_num": 0.0,
+            },
+            {
+                "clm_id": "W1",
+                "bene_id": "B1",
+                "nch_near_line_rec_ident_cd": "W",
+                "clm_from_dt": date(2023, 5, 1),
+                "clm_thru_dt": date(2023, 5, 1),
+            },
+            {
+                "clm_id": "Z1",
+                "bene_id": "B1",
+                "nch_near_line_rec_ident_cd": "Z",
+                "clm_from_dt": date(2023, 6, 1),
+                "clm_thru_dt": date(2023, 6, 1),
+            },
+            {
+                "clm_id": "T1",
+                "bene_id": "B1",
+                "clm_type_cd": "72",
+                "clm_from_dt": date(2023, 7, 1),
+                "clm_thru_dt": date(2023, 7, 1),
+            },
+        ]
+    )
+
+
+@pytest.fixture
+def v2_items() -> pl.DataFrame:
+    return _eob(
+        [
+            {
+                "clm_id": "V1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2023, 1, 1),
+                "clm_line_thru_dt": date(2023, 1, 5),
+                "clm_line_prod_rev_ctr_cd": "0301",
+                "clm_line_hcpcs_cd": "99213",
+                "clm_line_srvc_unit_qty": 1.0,
+                "clm_line_cvrd_pd_amt": 75.5,
+            },
+            {
+                "clm_id": "O1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2023, 2, 1),
+                "clm_line_thru_dt": date(2023, 2, 1),
+                "clm_line_hcpcs_cd": "99214",
+                "clm_pos_cd": "11",
+                "clm_line_srvc_unit_qty": 2.0,
+                "clm_line_cvrd_pd_amt": 120.0,
+                "clm_line_alowd_chrg_amt": 150.0,
+            },
+            {
+                "clm_id": "M1",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2023, 3, 1),
+                "clm_line_thru_dt": date(2023, 3, 1),
+                "clm_line_hcpcs_cd": "E1390",
+                "clm_pos_cd": "12",
+                "clm_line_cvrd_pd_amt": 50.0,
+                "clm_line_alowd_chrg_amt": 60.0,
+            },
+            {
+                "clm_id": "P2",
+                "clm_line_num": 1.0,
+                "clm_line_from_dt": date(2023, 4, 1),
+                "clm_line_thru_dt": date(2023, 4, 1),
+                "clm_line_ndc_cd": "12345678901",
+                "clm_line_srvc_unit_qty": 30.0,
+                "clm_line_cvrd_pd_amt": 10.0,
+                "clm_line_bene_pmt_amt": 5.0,
+            },
+        ]
+    )
+
+
+# ── helpers ──────────────────────────────────────────────────
+
+
+class TestHelpers:
+    def test_case_nests(self) -> None:
+        df = pl.DataFrame({"x": [1, 2, 3]})
+        expr = ex._case(
+            (nw.col("x") == 1, nw.lit("a")),
+            (nw.col("x") == 2, nw.lit("b")),
+            default=nw.lit("z"),
+        )
+        out = nw.from_native(df).select(expr.alias("y")).to_native()
+        assert out["y"].to_list() == ["a", "b", "z"]
+
+    def test_ensure_adds_typed_nulls(self) -> None:
+        df = nw.from_native(pl.DataFrame({"a": [1]}))
+        out = ex._ensure(df, {"a": nw.Int64, "b": nw.String, "c": nw.Date}).to_native()
+        assert out.columns == ["a", "b", "c"]
+        assert out.schema["b"] == pl.String and out.schema["c"] == pl.Date
+        assert ex._ensure(df, {"a": nw.Int64}).to_native().columns == ["a"]
+
+
+# ── stg_claim_header ─────────────────────────────────────────
+
+
+class TestStgClaimHeader:
+    def test_v3_routing_final_action_and_dedupe(self, v3_headers, xref) -> None:
+        out = stg_claim_header(
+            bcda__explanation_of_benefit=v3_headers, bcda___stg_beneficiary_xref=xref
+        )
+        fam = dict(
+            zip(out["clm_id"].to_list(), out["claim_family"].to_list(), strict=True)
+        )
+        assert fam == {
+            "I1": "institutional",
+            "C1": "professional",
+            "D1": "dme",
+            "P1": "pharmacy",
+            "N1": "institutional",  # NCH row wins the CCN-PAIR
+        }
+        assert "X1" not in fam  # NotFinalAction
+        assert "S1" not in fam  # SharedSystem twin dropped
+        # sparse fixture: every output column is in the contract; the runner's
+        # exact-match check is exercised on real tables in test_pipe_cclf
+        assert set(out.columns) <= set(BcdaStgClaimHeader.column_names())
+        assert {"claim_family", "current_bene_mbi_id"} <= set(out.columns)
+        mbi = dict(
+            zip(
+                out["clm_id"].to_list(),
+                out["current_bene_mbi_id"].to_list(),
+                strict=True,
+            )
+        )
+        assert mbi["I1"] == "MBI-1" and mbi["D1"] == "MBI-2"
+
+    def test_newest_wins_within_a_source(self, xref) -> None:
+        eob = _eob(
+            [
+                {
+                    "clm_id": "old",
+                    "clm_cntl_num": "C",
+                    "bene_id": "B1",
+                    "eob_type": "CARRIER",
+                    "system_type": "NationalClaimsHistory",
+                    "final_action": "FinalAction",
+                    "last_updated": "2026-01-01",
+                },
+                {
+                    "clm_id": "new",
+                    "clm_cntl_num": "C",
+                    "bene_id": "B1",
+                    "eob_type": "CARRIER",
+                    "system_type": "NationalClaimsHistory",
+                    "final_action": "FinalAction",
+                    "last_updated": "2026-02-01",
+                },
+            ]
+        )
+        out = stg_claim_header(
+            bcda__explanation_of_benefit=eob, bcda___stg_beneficiary_xref=xref
+        )
+        assert out["clm_id"].to_list() == ["new"]
+
+    def test_v2_routing_fallbacks(self, v2_headers, xref) -> None:
+        out = stg_claim_header(
+            bcda__explanation_of_benefit=v2_headers, bcda___stg_beneficiary_xref=xref
+        )
+        fam = dict(
+            zip(out["clm_id"].to_list(), out["claim_family"].to_list(), strict=True)
+        )
+        assert fam == {
+            "V1": "institutional",
+            "O1": "professional",
+            "M1": "dme",
+            "P2": "pharmacy",
+            "W1": "institutional",  # Part B institutional RIC
+            "Z1": None,  # unknown RIC → unrouted, kept for inspection
+            "T1": "professional",  # CLM-TYPE-CD without an EOB-TYPE
+        }
+        assert len(out) == 7  # nothing without a control number is collapsed
+
+    def test_pde_id_routes_pharmacy_without_codes(self, xref) -> None:
+        eob = _eob([{"clm_id": "Q", "pde_id": "Q", "bene_id": "B1"}])
+        out = stg_claim_header(
+            bcda__explanation_of_benefit=eob, bcda___stg_beneficiary_xref=xref
+        )
+        assert out["claim_family"].to_list() == ["pharmacy"]
+
+    def test_family_filters(self, v3_headers, xref) -> None:
+        hdr = stg_claim_header(
+            bcda__explanation_of_benefit=v3_headers, bcda___stg_beneficiary_xref=xref
+        )
+        assert sorted(
+            stg_part_a_header(bcda___stg_claim_header=hdr)["clm_id"].to_list()
+        ) == ["I1", "N1"]
+        assert stg_part_b_physician_header(bcda___stg_claim_header=hdr)[
+            "clm_id"
+        ].to_list() == ["C1"]
+        assert stg_part_b_dme_header(bcda___stg_claim_header=hdr)[
+            "clm_id"
+        ].to_list() == ["D1"]
+        assert stg_part_d_header(bcda___stg_claim_header=hdr)["clm_id"].to_list() == [
+            "P1"
+        ]
+
+
+# ── medical_claim ────────────────────────────────────────────
+
+
+@pytest.fixture
+def v3_medical(v3_headers, v3_items, xref) -> pl.DataFrame:
+    hdr = stg_claim_header(
+        bcda__explanation_of_benefit=v3_headers, bcda___stg_beneficiary_xref=xref
+    )
+    return medical_claim(
+        bcda___int_institutional_medical_claim=int_institutional_medical_claim(
+            bcda___stg_part_a_header=stg_part_a_header(bcda___stg_claim_header=hdr),
+            bcda__eob_item=v3_items,
+        ),
+        bcda___int_physician_medical_claim=int_physician_medical_claim(
+            bcda___stg_part_b_physician_header=stg_part_b_physician_header(
+                bcda___stg_claim_header=hdr
+            ),
+            bcda__eob_item=v3_items,
+        ),
+        bcda___int_dme_medical_claim=int_dme_medical_claim(
+            bcda___stg_part_b_dme_header=stg_part_b_dme_header(
+                bcda___stg_claim_header=hdr
+            ),
+            bcda__eob_item=v3_items,
+        ),
+    )
+
+
+class TestMedicalClaimV3:
+    def test_contract_and_rows(self, v3_medical) -> None:
+        assert set(v3_medical.columns) == set(BcdaMedicalClaim.column_names())
+        assert sorted(v3_medical["claim_id"].to_list()) == [
+            "C1",
+            "D1",
+            "I1",
+            "I1",
+            "N1",
+        ]
+        assert v3_medical.filter(pl.col("claim_id") == "X1").is_empty()
+
+    def test_institutional_line(self, v3_medical) -> None:
+        r = v3_medical.filter(
+            (pl.col("claim_id") == "I1") & (pl.col("claim_line_number") == 1)
+        ).to_dicts()[0]
+        assert r["claim_type"] == "institutional" and r["person_id"] == "MBI-1"
+        assert (r["admission_date"], r["discharge_date"]) == (
+            date(2024, 12, 31),
+            date(2025, 1, 9),
+        )
+        assert (
+            r["admit_source_code"],
+            r["admit_type_code"],
+            r["discharge_disposition_code"],
+        ) == ("7", "1", "01")
+        assert r["bill_type_code"] == "0111"  # NUBC, not "111111"
+        assert (r["drg_code_type"], r["drg_code"]) == ("ms-drg", "470")
+        assert r["revenue_center_code"] == "0120" and r["hcpcs_code"] == "99223"
+        assert r["service_unit_quantity"] == 1
+        assert (
+            r["rendering_npi"],
+            r["billing_npi"],
+            r["facility_npi"],
+            r["billing_tin"],
+        ) == ("ATT", "FAC", "FAC", "TIN-FAC")
+        assert r["rendering_tin"] is None and r["place_of_service_code"] is None
+        assert (
+            r["paid_amount"],
+            r["charge_amount"],
+            r["coinsurance_amount"],
+            r["deductible_amount"],
+        ) == (100.0, 250.0, 10.0, 5.0)
+        assert r["allowed_amount"] is None
+        assert r["paid_date"] == date(2025, 1, 20)  # NCH weekly process date
+        assert (
+            r["diagnosis_code_type"],
+            r["diagnosis_code_1"],
+            r["diagnosis_poa_1"],
+        ) == ("icd-10-cm", "I10", "Y")
+        assert (
+            r["procedure_code_type"],
+            r["procedure_code_1"],
+            r["procedure_date_1"],
+        ) == ("icd-10-pcs", "0SRC0J9", date(2025, 1, 2))
+        assert r["diagnosis_code_25"] is None and r["procedure_code_2"] is None
+        assert r["in_network_flag"] == 1 and r["data_source"] == "medicare bcda"
+        line2 = v3_medical.filter(
+            (pl.col("claim_id") == "I1") & (pl.col("claim_line_number") == 2)
+        ).to_dicts()[0]
+        assert line2["hcpcs_code"] is None and line2["charge_amount"] == 75.0
+
+    def test_professional_line(self, v3_medical) -> None:
+        r = v3_medical.filter(pl.col("claim_id") == "C1").to_dicts()[0]
+        assert r["claim_type"] == "professional"
+        assert r["place_of_service_code"] == "11" and r["hcpcs_modifier_1"] == "25"
+        assert (r["rendering_npi"], r["rendering_tin"], r["billing_npi"]) == (
+            "RND",
+            "TIN-RND",
+            "BLG",
+        )
+        assert (
+            r["facility_npi"] is None
+            and r["bill_type_code"] is None
+            and r["admission_date"] is None
+        )
+        assert (
+            r["paid_amount"],
+            r["allowed_amount"],
+            r["charge_amount"],
+            r["coinsurance_amount"],
+            r["deductible_amount"],
+        ) == (60.0, 75.0, 150.0, 15.0, 0.0)
+        assert r["paid_date"] == date(2025, 2, 10)  # CMS process date fallback
+        assert (
+            r["diagnosis_code_1"] == "E11.9"
+            and r["procedure_code_type"] is None
+            and r["diagnosis_poa_1"] is None
+        )
+
+    def test_dme_line(self, v3_medical) -> None:
+        r = v3_medical.filter(pl.col("claim_id") == "D1").to_dicts()[0]
+        assert r["claim_type"] == "professional" and r["person_id"] == "MBI-2"
+        assert (r["rendering_npi"], r["billing_npi"]) == (
+            "ORD",
+            "PAY",
+        )  # ordering / pay-to fallbacks
+        assert r["place_of_service_code"] == "12" and r["paid_date"] is None
+
+    def test_bill_type_fallback_from_v3_freq_code(self, xref) -> None:
+        """No NUBC coding captured → leading zero + the 3-char v3 freq code."""
+        eob = _eob(
+            [
+                {
+                    "clm_id": "H",
+                    "clm_cntl_num": "C",
+                    "bene_id": "B1",
+                    "eob_type": "HHA",
+                    "final_action": "FinalAction",
+                    "clm_bill_fac_type_cd": "3",
+                    "clm_bill_clsfctn_cd": "32",
+                    "clm_bill_freq_cd": "329",
+                },
+            ]
+        )
+        items = _eob([{"clm_id": "H", "clm_line_num": 1.0}])
+        hdr = stg_claim_header(
+            bcda__explanation_of_benefit=eob, bcda___stg_beneficiary_xref=xref
+        )
+        out = int_institutional_medical_claim(
+            bcda___stg_part_a_header=stg_part_a_header(bcda___stg_claim_header=hdr),
+            bcda__eob_item=items,
+        )
+        assert out["bill_type_code"].to_list() == ["0329"]
+
+    def test_pair_survivor_is_nch(self, v3_medical) -> None:
+        assert v3_medical.filter(pl.col("claim_id") == "N1")[
+            "hcpcs_code"
+        ].to_list() == ["G0463"]
+
+
+class TestMedicalClaimV2:
+    def test_v2_frames_map(self, v2_headers, v2_items, xref) -> None:
+        hdr = stg_claim_header(
+            bcda__explanation_of_benefit=v2_headers, bcda___stg_beneficiary_xref=xref
+        )
+        inst = int_institutional_medical_claim(
+            bcda___stg_part_a_header=stg_part_a_header(bcda___stg_claim_header=hdr),
+            bcda__eob_item=v2_items,
+        )
+        prof = int_physician_medical_claim(
+            bcda___stg_part_b_physician_header=stg_part_b_physician_header(
+                bcda___stg_claim_header=hdr
+            ),
+            bcda__eob_item=v2_items,
+        )
+        dme = int_dme_medical_claim(
+            bcda___stg_part_b_dme_header=stg_part_b_dme_header(
+                bcda___stg_claim_header=hdr
+            ),
+            bcda__eob_item=v2_items,
+        )
+        out = medical_claim(
+            bcda___int_institutional_medical_claim=inst,
+            bcda___int_physician_medical_claim=prof,
+            bcda___int_dme_medical_claim=dme,
+        )
+        assert set(out.columns) == set(BcdaMedicalClaim.column_names())
+        v1 = out.filter(pl.col("claim_id") == "V1").to_dicts()[0]
+        assert v1["bill_type_code"] == "111"  # v2 fac + clsfctn + freq
+        assert (
+            v1["drg_code"] == "470" and v1["charge_amount"] == 5000.0
+        )  # header total fallback
+        assert (v1["rendering_npi"], v1["billing_npi"]) == ("OP", "FACV")
+        assert (v1["admission_date"], v1["discharge_date"]) == (
+            date(2023, 1, 1),
+            date(2023, 1, 5),
+        )
+        assert v1["paid_date"] is None and v1["coinsurance_amount"] is None
+        o1 = out.filter(pl.col("claim_id") == "O1").to_dicts()[0]
+        assert (o1["rendering_npi"], o1["billing_npi"], o1["allowed_amount"]) == (
+            "RN",
+            "BL",
+            150.0,
+        )
+        m1 = out.filter(pl.col("claim_id") == "M1").to_dicts()[0]
+        assert (m1["rendering_npi"], m1["billing_npi"], m1["paid_amount"]) == (
+            "ORD",
+            "PAY",
+            50.0,
+        )
+        # W1 and T1 route but have no lines; Z1 is unrouted
+        assert sorted(set(out["claim_id"].to_list())) == ["M1", "O1", "V1"]
+
+
+# ── pharmacy_claim ───────────────────────────────────────────
+
+
+class TestPharmacyClaim:
+    def test_v3(self, v3_headers, v3_items, xref) -> None:
+        hdr = stg_claim_header(
+            bcda__explanation_of_benefit=v3_headers, bcda___stg_beneficiary_xref=xref
+        )
+        out = pharmacy_claim(
+            bcda___stg_part_d_header=stg_part_d_header(bcda___stg_claim_header=hdr),
+            bcda__eob_item=v3_items,
+        )
+        assert set(out.columns) == set(BcdaPharmacyClaim.column_names())
+        r = out.to_dicts()[0]
+        assert (r["claim_id"], r["claim_line_number"], r["person_id"]) == (
+            "P1",
+            1,
+            "MBI-2",
+        )
+        assert (r["prescribing_provider_npi"], r["dispensing_provider_npi"]) == (
+            "PRS",
+            "PHARM",
+        )
+        assert (r["dispensing_date"], r["paid_date"]) == (
+            date(2025, 4, 1),
+            date(2025, 4, 3),
+        )
+        assert (r["ndc_code"], r["quantity"], r["days_supply"], r["refills"]) == (
+            "00002143380",
+            30,
+            30,
+            2,
+        )
+        assert (
+            r["paid_amount"] is None
+        )  # sandbox emits no plan-paid; header clm_pmt_amt is NULL too
+        assert (r["charge_amount"], r["copayment_amount"]) == (
+            118.0,
+            7.5,
+        )  # line drug cost, header paid-by-patient
+        assert r["allowed_amount"] is None and r["deductible_amount"] is None
+
+    def test_v2(self, v2_headers, v2_items, xref) -> None:
+        hdr = stg_claim_header(
+            bcda__explanation_of_benefit=v2_headers, bcda___stg_beneficiary_xref=xref
+        )
+        out = pharmacy_claim(
+            bcda___stg_part_d_header=stg_part_d_header(bcda___stg_claim_header=hdr),
+            bcda__eob_item=v2_items,
+        )
+        r = out.to_dicts()[0]
+        assert r["claim_id"] == "PDE-2" and r["dispensing_provider_npi"] == "DSP2"
+        assert (r["quantity"], r["days_supply"], r["refills"]) == (30, 30, 0)
+        assert (r["paid_amount"], r["copayment_amount"], r["charge_amount"]) == (
+            10.0,
+            5.0,
+            None,
+        )
+        assert r["paid_date"] == date(2023, 4, 1)
+
+
+# ── eligibility ──────────────────────────────────────────────
+
+
+def _patients() -> pl.DataFrame:
+    return pl.DataFrame(
+        {
+            "bene_id": ["B1", "B2", "B3"],
+            "bene_mbi_id": ["M1", "M2", "M3"],
+            "bene_sex_cd": ["1", "female", "9"],
+            "bene_race_cd": ["1", "5", None],
+            "bene_dob": [date(1950, 1, 1), date(1960, 2, 2), None],
+            "bene_death_dt": [None, date(2025, 1, 1), None],
+            "bene_1st_name": ["a", "b", "c"],
+            "bene_midl_name": ["x", None, None],
+            "bene_last_name": ["A", "B", "C"],
+            "bene_line_1_adr": ["1 st", None, None],
+            "geo_zip_plc_name": ["Town", None, None],
+            "geo_usps_state_cd": ["TX", "VA", None],
+            "geo_zip5_cd": ["77550", None, None],
+        }
+    )
+
+
+class TestEligibility:
+    def test_v3_coverage_rows(self) -> None:
+        coverage = pl.DataFrame(
+            {
+                "subscriber_id": ["M1", "M1", "M1", "M2", "M2", "M3"],
+                "class_plan": [
+                    "Part C",
+                    "Part B",
+                    "Part A",
+                    "Part B",
+                    "Dual Medicare/Medicaid",
+                    "Part A",
+                ],
+                "status": [
+                    "active",
+                    "active",
+                    "cancelled",
+                    "active",
+                    "active",
+                    "active",
+                ],
+                "ms_cd": [None, "10", "20", "10", None, "30"],
+                "orec": [None, "0", "1", "0", None, "2"],
+                "dual_stus_cd": [None, None, None, None, "08", None],
+                "bene_part_a_enrlmt_bgn_dt": [
+                    None,
+                    None,
+                    date(2020, 1, 1),
+                    None,
+                    None,
+                    date(2021, 1, 1),
+                ],
+                "mdcr_covg_a_end_dt": [
+                    None,
+                    None,
+                    date(2024, 12, 31),
+                    None,
+                    None,
+                    None,
+                ],
+                "bene_part_b_enrlmt_bgn_dt": [
+                    None,
+                    date(2020, 6, 1),
+                    None,
+                    date(2022, 1, 1),
+                    None,
+                    None,
+                ],
+                "mdcr_covg_b_end_dt": [None, None, None, None, None, None],
+            }
+        )
+        demo = stg_beneficiary_demographics(
+            bcda__patient=_patients(), bcda__coverage=coverage
+        ).sort("bene_mbi_id")
+        assert set(demo.columns) == set(BcdaStgBeneficiaryDemographics.column_names())
+        assert demo["coverage_plan"].to_list() == ["Part A", "Part B", "Part A"]
+        assert demo["ms_cd"].to_list() == ["20", "10", "30"]
+        assert demo["dual_stus_cd_dual"].to_list() == [None, "08", None]
+        assert demo["enrollment_start_date"].to_list() == [
+            date(2020, 1, 1),
+            date(2022, 1, 1),
+            date(2021, 1, 1),
+        ]
+        elig = eligibility(bcda___stg_beneficiary_demographics=demo).sort("person_id")
+        assert set(elig.columns) == set(BcdaEligibility.column_names())
+        assert elig["gender"].to_list() == ["male", "female", "unknown"]
+        assert elig["race"].to_list() == ["white", "hispanic", "unknown"]
+        assert elig["ethnicity"].to_list() == [None, "hispanic", None]
+        assert elig["enrollment_status"].to_list() == ["inactive", "active", "active"]
+        assert elig["dual_status_code"].to_list() == [None, "08", None]
+        assert elig["medicare_status_code"].to_list() == ["20", "10", "30"]
+        assert elig["original_reason_entitlement_code"].to_list() == ["1", "0", "2"]
+        assert elig["enrollment_end_date"].to_list() == [date(2024, 12, 31), None, None]
+        assert elig["death_flag"].to_list() == [0, 1, 0]
+        assert (
+            elig["payer_type"].to_list() == ["medicare"] * 3
+            and elig["hospice_flag"].null_count() == 3
+        )
+
+    def test_v2_coverage_without_plan_columns(self) -> None:
+        coverage = pl.DataFrame(
+            {
+                "subscriber_id": ["M1", "M3"],
+                "ms_cd": ["10", "30"],
+                "orec": ["0", "2"],
+                "dual_01": ["04", None],
+                "bene_part_a_enrlmt_bgn_dt": [date(2019, 1, 1), None],
+                "mdcr_covg_a_end_dt": [None, None],
+            }
+        )
+        demo = stg_beneficiary_demographics(
+            bcda__patient=_patients(), bcda__coverage=coverage
+        ).sort("bene_mbi_id")
+        assert demo["coverage_plan"].to_list() == [None, None, None]
+        elig = eligibility(bcda___stg_beneficiary_demographics=demo).sort("person_id")
+        assert elig["dual_status_code"].to_list() == ["04", None, None]
+        assert elig["medicare_status_code"].to_list() == ["10", None, "30"]
+        assert elig["enrollment_status"].null_count() == 3
+        assert elig["enrollment_start_date"].to_list() == [date(2019, 1, 1), None, None]

tests/bcda/test_pipe_cclf.py

@@ -9,21 +9,22 @@ import pytest
 
 class TestBuildPipeline:
     def test_build_returns_pipeline(self) -> None:
-        """_build_pipeline() returns a Pipeline with 12 exprs."""
+        """_build_pipeline() returns a Pipeline with 13 exprs."""
         from bcda.pipe.cclf import _build_pipeline
 
         pipeline = _build_pipeline()
         assert hasattr(pipeline, "exprs")
-        assert len(pipeline.exprs) == 12
+        assert len(pipeline.exprs) == 13
         assert hasattr(pipeline, "run")
 
     def test_expr_names(self) -> None:
-        """All 12 expected expression names are present."""
+        """All 13 expected expression names are present."""
         from bcda.pipe.cclf import _build_pipeline
 
         pipeline = _build_pipeline()
         names = [e.name for e in pipeline.exprs]
         assert "bcda._stg_beneficiary_xref" in names
+        assert "bcda._stg_claim_header" in names
         assert "bcda._stg_part_a_header" in names
         assert "bcda._int_institutional_medical_claim" in names
         assert "bcda._stg_part_b_physician_header" in names
@@ -52,7 +53,7 @@ class TestLazyGetattr:
 
         pipeline = cclf_mod.pipeline
         assert hasattr(pipeline, "exprs")
-        assert len(pipeline.exprs) == 12
+        assert len(pipeline.exprs) == 13
 
     def test_run_attr(self) -> None:
         """Accessing .run triggers lazy init and returns callable."""
@@ -76,3 +77,78 @@ class TestLazyGetattr:
             match="has no attribute 'nonexistent'",
         ):
             _ = cclf_mod.nonexistent
+
+
+class TestRunOnOfficialSamples:
+    """The whole connector against CMS's v3 sample files through DuckDB.
+
+    Every expression's output is checked against its contract by the
+    runner, so this is the proof that the v3-flattened tables carry
+    what the connector reads and that the outputs are input_layer-shaped.
+    """
+
+    def test_end_to_end(self, tmp_path) -> None:
+        import shutil
+        from pathlib import Path
+
+        import duckdb
+
+        from aco.lake.context import DuckDBContext
+        from aco.load.bcda import load_bcda
+        from bcda.pipe.cclf import pipeline
+
+        samples = Path(__file__).resolve().parents[2] / "dev" / "vendor" / "bcda-site"
+        src = tmp_path / "ndjson"
+        src.mkdir()
+        for rtype in ("Patient", "Coverage", "ExplanationOfBenefit"):
+            shutil.copy(samples / f"v3_{rtype}.ndjson", src / f"{rtype}.ndjson")
+        db = tmp_path / "aco.duckdb"
+        from unittest.mock import patch
+
+        with patch("conf.path", side_effect=lambda k: tmp_path / "store"):
+            load_bcda(ndjson_dir=src, database=str(db))
+
+        ctx = DuckDBContext(database=str(db), read_only=True)
+        cache = pipeline.run(ctx.load)
+        assert set(cache) >= {
+            "bcda._stg_claim_header",
+            "bcda.medical_claim",
+            "bcda.pharmacy_claim",
+            "bcda.eligibility",
+        }
+        header = cache["bcda._stg_claim_header"]
+        # 36 NotFinalAction sample claims are dropped; the rest are routed
+        assert len(header) == 233
+        assert set(header["claim_family"].to_list()) == {
+            "institutional",
+            "professional",
+            "dme",
+            "pharmacy",
+        }
+        medical = cache["bcda.medical_claim"]
+        assert medical["claim_id"].n_unique() == 131
+        assert set(medical["claim_type"].to_list()) == {"institutional", "professional"}
+        inst = medical.filter(medical["claim_type"] == "institutional")
+        assert inst["bill_type_code"].null_count() == 0
+        assert (
+            inst["bill_type_code"].str.len_chars().max() == 4
+        )  # NUBC, not fac+clsfctn+freq
+        assert inst["admission_date"].null_count() == 0
+        assert inst["paid_amount"].null_count() == 0
+        assert inst["deductible_amount"].null_count() == 0
+        prof = medical.filter(medical["claim_type"] == "professional")
+        assert prof["place_of_service_code"].null_count() < len(prof) * 0.05
+        assert prof["rendering_npi"].null_count() == 0
+        assert len(cache["bcda.pharmacy_claim"]) == 102
+        rx = cache["bcda.pharmacy_claim"]
+        assert rx["ndc_code"].null_count() == 0 and rx["days_supply"].null_count() == 0
+        assert rx["charge_amount"].null_count() == 0  # total drug cost
+        elig = cache["bcda.eligibility"]
+        assert len(elig) == 94
+        assert set(elig["gender"].to_list()) == {"male", "female"}
+        assert elig["medicare_status_code"].null_count() < len(elig)
+        assert set(elig["enrollment_status"].drop_nulls().to_list()) <= {
+            "active",
+            "inactive",
+        }
+        duckdb.connect(str(db), read_only=True).close()

tests/bcda/test_v3.py

@@ -1061,6 +1061,7 @@ class TestFlattenEobV3:
             h.clm_srvc_clsfctn_type_cd,
             h.clm_bill_freq_cd,
         ) == ("1", "1", "1", "1")
+        assert h.clm_bill_type_cd == "0111"
 
     def test_c4bb_code_falls_back_to_any_system(self) -> None:
         r = _eob(

tests/cli/test_cli_run.py

@@ -143,6 +143,18 @@ class TestLoadCommands:
             result = runner.invoke(app, ["load", "bcda", "--ndjson-dir", "/tmp/fake"])
             assert result.exit_code == 0
 
+    def test_bcda_run_pipeline_flag(self) -> None:
+        with patch(
+            "aco.load.bcda.load_bcda",
+            return_value={"bcda.patient": 50, "input_layer.eligibility": 50},
+        ) as load:
+            result = runner.invoke(
+                app, ["load", "bcda", "--ndjson-dir", "/tmp/fake", "--run-pipeline"]
+            )
+            assert result.exit_code == 0
+            assert load.call_args.kwargs["run_pipeline"] is True
+            assert "input_layer.eligibility: 50 rows" in result.output
+
     def test_bcda_not_found(self) -> None:
         with patch("aco.load.bcda.load_bcda", side_effect=FileNotFoundError("missing")):
             result = runner.invoke(app, ["load", "bcda", "--ndjson-dir", "/tmp/fake"])

tests/test_dab_generator.py

@@ -169,7 +169,8 @@ class TestPythonResources:
 
     def test_schemas_count(self):
         r = self._resources()
-        assert len(r.schemas) == 14, f"Expected 14 schemas, got {len(r.schemas)}"
+        # 14 aco pipelines + the bcda connector registered when stack[bcda] is installed
+        assert len(r.schemas) == 15, f"Expected 15 schemas, got {len(r.schemas)}"
 
     def test_schemas_include_core(self):
         r = self._resources()