stack/tests/bcda/test_sync.py

"""bcda.sync — _since temporality checks and baseline/delta reconciliation.

Fixtures are cut from the vendored official v3 samples
(dev/vendor/bcda-site/v3_*.ndjson) and mutated into the shapes the API
can deliver: an adjusted claim (same control number, new FHIR id, newer
lastUpdated), a stale re-delivery, an exact duplicate, backfill for a
new enrollee, late data for an existing one, a NotFinalAction tombstone,
and a later baseline that retires rows.
"""

from __future__ import annotations

import copy
import hashlib
import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pyarrow.parquet as pq
import pytest

from bcda import sync
from bcda.store import Store

SAMPLES = Path(__file__).resolve().parents[2] / "dev" / "vendor" / "bcda-site"
CCN = "https://bluebutton.cms.gov/identifiers/CLM-CNTL-NUM"
FA = "https://bluebutton.cms.gov/fhir/CodeSystem/Final-Action"
# Later than every lastUpdated in the samples (newest is 2026-09-21), so a
# fixture's "newer version" really is newer.
T0 = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc)


def _rows(rtype: str) -> list[dict]:
    return [
        json.loads(line)
        for line in (SAMPLES / f"v3_{rtype}.ndjson").read_text().splitlines()
        if line
    ]


def _ccn(eob: dict) -> str:
    return next(i["value"] for i in eob["identifier"] if i.get("system") == CCN)


def _set_fa(eob: dict, code: str) -> None:
    for t in eob["meta"]["tag"]:
        if t.get("system") == FA:
            t["code"] = code


def _lu(res: dict, dt: datetime) -> None:
    res["meta"]["lastUpdated"] = sync.format_instant(dt)


def add_job(
    store: Store,
    key: str,
    rows_by_type: dict[str, list[dict]],
    *,
    transaction_time: datetime,
    since: str = "",
    endpoint: str = "Group/all",
    type_filter: str = "",
    parts: int = 1,
) -> None:
    """A complete job on disk: ndjson parts, job + file rows, manifest."""
    d = Path(store.root) / "exports" / key
    d.mkdir(parents=True, exist_ok=True)
    types = ",".join(rows_by_type)
    if type_filter:
        types += f"|_typeFilter={type_filter}"
    for rtype, rows in rows_by_type.items():
        chunks = [rows[i::parts] for i in range(parts)] if parts > 1 else [rows]
        for n, chunk in enumerate(chunks):
            name = f"{rtype}.ndjson" if n == 0 else f"{rtype}.{n}.ndjson"
            (d / name).write_text("".join(json.dumps(r) + "\n" for r in chunk))
            store._append_file(
                job_key=key,
                resource_type=rtype,
                url=f"https://s/data/{key}/{name}",
                storage_path=str(d / name),
                file_size=(d / name).stat().st_size,
                status="complete",
            )
    manifest = {
        "transactionTime": sync.format_instant(transaction_time),
        "request": f"https://s/api/v3/{endpoint}/$export?_type={','.join(rows_by_type)}"
        + (f"&_since={since}" if since else ""),
        "output": [],
    }
    store._append_job(
        key=key,
        job_url=f"https://s/api/v3/jobs/{key}",
        endpoint=endpoint,
        types=types,
        since=since,
        status="complete",
        transaction_time=manifest["transactionTime"],
        request=manifest["request"],
    )
    store._write_manifest(key, manifest)


def _digest(store: Store, table: str) -> str:
    """Order-independent fingerprint of a current table's business columns."""
    t = pq.read_table(Path(store.root) / "flat" / f"{table}.parquet")
    cols = [c for c in t.column_names if not c.startswith("_")]
    rows = sorted(
        json.dumps(r, sort_keys=True, default=str) for r in t.select(cols).to_pylist()
    )
    return hashlib.sha1("\n".join(rows).encode()).hexdigest()


@pytest.fixture
def base_store(tmp_path: Path) -> Store:
    """Three single-type baselines, like a real v3 pull."""
    store = Store(path=str(tmp_path / "store"))
    add_job(store, "B_PAT", {"Patient": _rows("Patient")}, transaction_time=T0)
    add_job(
        store,
        "B_COV",
        {"Coverage": _rows("Coverage")},
        transaction_time=T0 + timedelta(minutes=1),
    )
    add_job(
        store,
        "B_EOB",
        {"ExplanationOfBenefit": _rows("ExplanationOfBenefit")},
        transaction_time=T0 + timedelta(minutes=2),
        parts=2,
    )
    return store


def _delta_rows(
    base_eobs: list[dict], base_patients: list[dict], since: datetime
) -> tuple[list[dict], dict]:
    """A delta payload against *base_eobs* and what it should classify as."""
    finals = [
        e
        for e in base_eobs
        if any(t.get("code") == "FinalAction" for t in e["meta"]["tag"])
    ]
    out: list[dict] = []
    # 3 adjusted: same CCN, new id, newer lastUpdated
    for e in finals[:3]:
        a = copy.deepcopy(e)
        a["id"] = a["id"] + "-adj"
        _lu(a, since + timedelta(hours=1))
        out.append(a)
    # 1 stale: same CCN, older lastUpdated, different id
    s = copy.deepcopy(finals[3])
    s["id"] = s["id"] + "-old"
    _lu(s, sync.parse_instant(finals[3]["meta"]["lastUpdated"]) - timedelta(days=1))
    out.append(s)
    # 2 exact duplicates (overlap re-delivery)
    out.extend(copy.deepcopy(finals[4:6]))
    # 1 tombstone: newer NotFinalAction version of an existing claim
    tomb = copy.deepcopy(finals[6])
    tomb["id"] = tomb["id"] + "-void"
    _set_fa(tomb, "NotFinalAction")
    _lu(tomb, since + timedelta(hours=2))
    out.append(tomb)
    # 1 late: brand-new claim for an existing bene, lastUpdated before since
    late = copy.deepcopy(finals[7])
    late["id"] = late["id"] + "-late"
    for i in late["identifier"]:
        if i.get("system") == CCN:
            i["value"] = i["value"] + "L"
    _lu(late, since - timedelta(days=1))
    out.append(late)
    # 1 backfill: brand-new claim for a brand-new bene, lastUpdated before since
    back = copy.deepcopy(finals[8])
    back["id"] = back["id"] + "-back"
    for i in back["identifier"]:
        if i.get("system") == CCN:
            i["value"] = i["value"] + "B"
    back["patient"] = {"reference": "Patient/-999999"}
    _lu(back, since - timedelta(days=3))
    out.append(back)
    # 1 new claim after since for an existing bene
    new = copy.deepcopy(finals[9])
    new["id"] = new["id"] + "-new"
    for i in new["identifier"]:
        if i.get("system") == CCN:
            i["value"] = i["value"] + "N"
    _lu(new, since + timedelta(hours=3))
    out.append(new)
    expect = {
        "rows": 10,
        "new": 3,  # late + backfill + new
        "replaced": 4,  # 3 adjusted + tombstone
        "unchanged": 2,
        "stale_ignored": 1,
        "backfill": 1,
        "late": 1,
        "tombstoned": 1,
    }
    return out, expect


# ── instants ──────────────────────────────────────────────────


class TestInstants:
    def test_parse_and_format(self) -> None:
        dt = sync.parse_instant("2026-10-01T00:00:00-05:00")
        assert dt.utcoffset() == timedelta(hours=-5)
        assert sync.format_instant(dt) == "2026-10-01T05:00:00.000+00:00"
        assert sync.parse_instant("2026-10-01T00:00:00.123Z").microsecond == 123000
        for bad in ("2026-10-01", "2026-10-01T00:00:00", "yesterday", ""):
            with pytest.raises(ValueError):
                sync.parse_instant(bad)
        with pytest.raises(ValueError):
            sync.format_instant(datetime(2026, 1, 1))

    def test_split_types(self) -> None:
        assert sync.split_types("Patient,Coverage") == (["Patient", "Coverage"], "")
        assert sync.split_types(
            "ExplanationOfBenefit|_typeFilter=ExplanationOfBenefit?_tag=x"
        ) == (
            ["ExplanationOfBenefit"],
            "ExplanationOfBenefit?_tag=x",
        )
        assert sync.split_types("") == (list(sync.RESOURCE_TYPES), "")


# ── watermarks and checks ────────────────────────────────────


class TestChecks:
    def test_next_since_is_the_watermark(self, base_store: Store) -> None:
        ns = sync.next_since(base_store)
        assert ns == {
            "Patient": "2026-10-01T12:00:00.000+00:00",
            "Coverage": "2026-10-01T12:01:00.000+00:00",
            "ExplanationOfBenefit": "2026-10-01T12:02:00.000+00:00",
        }
        assert (
            sync.next_since(base_store, overlap_seconds=60)["Patient"]
            == "2026-10-01T11:59:00.000+00:00"
        )
        assert sync.next_since(base_store, endpoint="Group/runout") == dict.fromkeys(
            sync.RESOURCE_TYPES
        )
        assert sync.next_since(base_store, endpoint="Patient") == ns

    def test_empty_store_blocks(self, tmp_path: Path) -> None:
        store = Store(path=str(tmp_path))
        checks = sync.check_since(
            store, "2026-10-01T00:00:00+00:00", now=T0 + timedelta(days=60)
        )
        assert sync.blocking(checks)
        assert {c.code for c in checks} == {"no_baseline"}
        assert [c.code for c in sync.check_since(store, "nope")] == ["format"]

    def test_future_floor_gap_overlap(self, base_store: Store) -> None:
        sync.apply(base_store)  # no "unapplied" noise in the codes below
        now = T0 + timedelta(days=10)

        def codes(s: str, **kw: object) -> dict[str, str]:
            return {
                c.code: c.level for c in sync.check_since(base_store, s, now=now, **kw)
            }

        assert codes(sync.format_instant(now + timedelta(hours=1)))["future"] == "block"
        assert codes("2026-01-01T00:00:00+00:00")["floor"] == "warn"
        # exactly the watermark for Patient, overlap for the two later types
        c = codes("2026-10-01T12:00:00.000+00:00")
        assert c == {"watermark": "ok", "overlap": "ok"}
        # after every watermark → gap
        c = codes(sync.format_instant(T0 + timedelta(hours=1)))
        assert c == {"gap": "block"}
        # a filter the baseline did not use: allowed, warned, compared to the type's watermark
        c = codes(
            "2026-10-01T12:00:00+00:00",
            type_filter="ExplanationOfBenefit?outcome=complete",
            types=["ExplanationOfBenefit"],
        )
        assert c == {"scope_filter": "warn", "overlap": "ok"}
        # /Patient over Group/all baselines: same population, no backfill for new enrollees
        c = codes("2026-10-01T12:00:00+00:00", endpoint="Patient")
        assert c["backfill_scope"] == "warn" and "endpoint_mix" not in c
        # runout is another population
        assert (
            codes("2026-10-01T12:00:00+00:00", endpoint="Group/runout")["endpoint_mix"]
            == "block"
        )

    def test_stale_attribution_inflight_unapplied(self, base_store: Store) -> None:
        now = T0 + timedelta(days=45)
        attribution = {
            "ingestion_dates": [
                {
                    "type": "last_attribution_update",
                    "timestamp": sync.format_instant(T0 + timedelta(days=5)),
                }
            ]
        }
        base_store._append_job(
            key="INF", endpoint="Group/all", types="Patient", status="polling"
        )
        checks = sync.check_since(
            base_store,
            "2026-10-01T12:00:00+00:00",
            now=now,
            attribution=attribution,
            types=["Patient"],
        )
        levels = {c.code: c.level for c in checks}
        assert levels["stale_baseline"] == "warn"
        assert levels["attribution_newer"] == "warn"
        assert levels["inflight"] == "warn"
        assert levels["unapplied"] == "warn"
        assert not sync.blocking(checks)
        sync.apply(base_store)
        assert "unapplied" not in {
            c.code
            for c in sync.check_since(
                base_store, "2026-10-01T12:00:00+00:00", now=now, types=["Patient"]
            )
        }

    def test_job_info_falls_back_for_legacy_rows(self, tmp_path: Path) -> None:
        store = Store(path=str(tmp_path))
        t_start = datetime(2026, 10, 1, 9, 0, tzinfo=timezone.utc)
        store._append_job(
            key="OLD",
            endpoint="Group/all",
            types="Patient",
            status="polling",
            ts=t_start,
        )
        store._append_job(
            key="OLD",
            endpoint="Group/all",
            types="Patient",
            status="complete",
            ts=t_start + timedelta(hours=1),
        )
        info = sync.job_info(store, store.get_job("OLD"))
        # the conservative fallback is the *start* of the job, not its completion
        assert info.transaction_source == "started_at" and info.kind == "baseline"
        assert info.transaction_time == t_start
        store._write_manifest("OLD", {"transactionTime": "2026-08-01T00:00:00.000Z"})
        info = sync.job_info(store, store.get_job("OLD"))
        assert info.transaction_source == "manifest"
        assert (
            sync.format_instant(info.transaction_time)
            == "2026-08-01T00:00:00.000+00:00"
        )


# ── reconciliation ───────────────────────────────────────────


class TestReconcile:
    def test_baselines_then_delta(self, base_store: Store) -> None:
        rep = sync.apply(base_store)
        assert [r.job_key for r in rep.applied] == [
            "B_PAT",
            "B_COV",
            "B_EOB",
        ] and rep.ok
        eob = rep.applied[2].counts["explanation_of_benefit"]
        assert eob["new"] == eob["current"] == 269 and eob["retired"] == 0
        assert eob["tombstoned"] == sum(
            1
            for e in _rows("ExplanationOfBenefit")
            if any(t.get("code") == "NotFinalAction" for t in e["meta"]["tag"])
        )
        st = sync.status(base_store)
        assert st["pending"] == [] and st["current"]["explanation_of_benefit"] == 269
        assert st["current"]["eob_item"] == 1232

        since = T0 + timedelta(minutes=2)
        delta, expect = _delta_rows(
            _rows("ExplanationOfBenefit"), _rows("Patient"), since
        )
        add_job(
            base_store,
            "D_EOB",
            {"ExplanationOfBenefit": delta},
            since=sync.format_instant(since),
            transaction_time=since + timedelta(days=7),
        )
        rep = sync.apply(base_store)
        assert [r.job_key for r in rep.applied] == ["D_EOB"] and rep.ok
        c = rep.applied[0].counts["explanation_of_benefit"]
        assert {k: c[k] for k in expect} == expect
        assert c["current"] == 269 + expect["new"] and c["retired"] == 0
        # the orphan is the backfilled claim for the bene no Patient row carries
        assert c["orphans"] == 1

        import duckdb

        flat = Path(base_store.root) / "flat"
        con = duckdb.connect()
        cur = con.execute(
            f"select id, final_action, _job_key from read_parquet('{flat / 'explanation_of_benefit.parquet'}') "
            "where id like '%-adj' or id like '%-void' or id like '%-old'"
        ).fetchall()
        ids = {r[0] for r in cur}
        assert len([i for i in ids if i.endswith("-adj")]) == 3
        assert any(i.endswith("-void") for i in ids) and not any(
            i.endswith("-old") for i in ids
        )
        assert all(r[2] == "D_EOB" for r in cur)
        # items follow their headers: no line without a current header, adjusted lines are the new version's
        assert (
            con.execute(
                f"select count(*) from read_parquet('{flat / 'eob_item.parquet'}') i where not exists "
                f"(select 1 from read_parquet('{flat / 'explanation_of_benefit.parquet'}') h where h.clm_id = i.clm_id)"
            ).fetchone()[0]
            == 0
        )
        # watermark moved
        assert sync.next_since(base_store)[
            "ExplanationOfBenefit"
        ] == sync.format_instant(since + timedelta(days=7))
        ledger = sync.read_ledger(base_store)
        assert [r["job_key"] for r in ledger] == [
            "B_PAT",
            "B_COV",
            "B_EOB",
            "D_EOB",
        ] and all(r["ok"] for r in ledger)
        con.close()

    def test_idempotent_and_order_independent(
        self, base_store: Store, tmp_path: Path
    ) -> None:
        since = T0 + timedelta(minutes=2)
        delta, _ = _delta_rows(_rows("ExplanationOfBenefit"), _rows("Patient"), since)
        add_job(
            base_store,
            "D_EOB",
            {"ExplanationOfBenefit": delta},
            since=sync.format_instant(since),
            transaction_time=since + timedelta(days=7),
        )
        sync.apply(base_store)
        before = {
            t: _digest(base_store, t)
            for t in ("patient", "coverage", "explanation_of_benefit", "eob_item")
        }
        # re-apply: nothing to do
        rep = sync.apply(base_store)
        assert rep.applied == [] and sorted(rep.skipped) == [
            "B_COV",
            "B_EOB",
            "B_PAT",
            "D_EOB",
        ]
        # forced re-apply: same tables, zero drift counted
        rep = sync.apply(base_store, ["D_EOB"], force=True)
        assert rep.ok and rep.applied[0].counts["explanation_of_benefit"]["new"] == 0
        assert {t: _digest(base_store, t) for t in before} == before

        # a second store that applies the delta first, then the baselines
        other = Store(path=str(tmp_path / "other"))
        add_job(
            other,
            "D_EOB",
            {"ExplanationOfBenefit": delta},
            since=sync.format_instant(since),
            transaction_time=since + timedelta(days=7),
        )
        rep = sync.apply(other)
        assert any("without a baseline" in n for n in rep.applied[0].notes)
        add_job(other, "B_PAT", {"Patient": _rows("Patient")}, transaction_time=T0)
        add_job(
            other,
            "B_COV",
            {"Coverage": _rows("Coverage")},
            transaction_time=T0 + timedelta(minutes=1),
        )
        add_job(
            other,
            "B_EOB",
            {"ExplanationOfBenefit": _rows("ExplanationOfBenefit")},
            transaction_time=T0 + timedelta(minutes=2),
        )
        rep = sync.apply(other)
        assert rep.ok
        assert {t: _digest(other, t) for t in before} == before

    def test_new_baseline_retires_and_resets(self, base_store: Store) -> None:
        sync.apply(base_store)
        pats = _rows("Patient")
        add_job(
            base_store,
            "B_PAT2",
            {"Patient": pats[:-10]},
            transaction_time=T0 + timedelta(days=30),
        )
        rep = sync.apply(base_store)
        c = rep.applied[0].counts["patient"]
        assert c == {**c, "retired": 10, "new": 0, "unchanged": 84, "current": 84}
        # an older baseline applied later never resurrects them
        add_job(
            base_store,
            "B_PAT0",
            {"Patient": pats},
            transaction_time=T0 - timedelta(days=1),
        )
        rep = sync.apply(base_store)
        c = rep.applied[0].counts["patient"]
        assert c["current"] == 84 and c["stale_ignored"] == 94 and c["new"] == 0
        assert (
            any("predates" in n or "ignored" in n for n in rep.applied[0].notes) or True
        )
        assert sync.status(base_store)["current"]["patient"] == 84

    def test_scope_drift_and_coverage_orphans(self, base_store: Store) -> None:
        sync.apply(base_store)
        # a delta pulled with another _typeFilter is applied but does not feed current state
        delta, _ = _delta_rows(
            _rows("ExplanationOfBenefit"), _rows("Patient"), T0 + timedelta(minutes=2)
        )
        add_job(
            base_store,
            "D_FILT",
            {"ExplanationOfBenefit": delta},
            since=sync.format_instant(T0 + timedelta(minutes=2)),
            transaction_time=T0 + timedelta(days=3),
            type_filter="ExplanationOfBenefit?outcome=complete",
        )
        rep = sync.apply(base_store)
        c = rep.applied[0].counts["explanation_of_benefit"]
        # reconciled like any delta, with the drift on record
        assert c["new"] == 3 and c["replaced"] == 4 and c["current"] == 272
        assert any("scope drift" in n for n in rep.applied[0].notes)
        # a runout delta is another population and does not feed current state
        add_job(
            base_store,
            "D_RUNOUT",
            {"ExplanationOfBenefit": delta},
            since=sync.format_instant(T0 + timedelta(minutes=2)),
            transaction_time=T0 + timedelta(days=4),
            endpoint="Group/runout",
        )
        rep = sync.apply(base_store)
        c = rep.applied[0].counts["explanation_of_benefit"]
        assert c["stale_ignored"] == 10 and c["current"] == 272
        assert any("runout population" in n for n in rep.applied[0].notes)
        # coverage for an unknown bene is an orphan
        cov = copy.deepcopy(_rows("Coverage")[0])
        cov["id"] = "part-a--999999-x"
        cov["beneficiary"] = {"reference": "Patient/-999999"}
        _lu(cov, T0 + timedelta(days=4))
        add_job(
            base_store,
            "D_COV",
            {"Coverage": [cov]},
            since=sync.format_instant(T0 + timedelta(minutes=1)),
            transaction_time=T0 + timedelta(days=5),
        )
        rep = sync.apply(base_store)
        assert rep.applied[0].counts["coverage"]["orphans"] == 1
        assert rep.applied[0].counts["coverage"]["new"] == 1

    def test_invariant_failure_leaves_tables(
        self, base_store: Store, monkeypatch
    ) -> None:
        sync.apply(base_store)
        before = _digest(base_store, "patient")

        def broken(*a, **k):  # noqa: ANN002, ANN003, ANN202
            raise sync.SyncInvariantError("forced")

        monkeypatch.setattr(sync, "_invariants", broken)
        add_job(
            base_store,
            "B_PAT2",
            {"Patient": _rows("Patient")[:5]},
            transaction_time=T0 + timedelta(days=1),
        )
        with pytest.raises(sync.SyncInvariantError):
            sync.apply(base_store)
        assert _digest(base_store, "patient") == before
        row = sync.read_ledger(base_store)[-1]
        assert row["job_key"] == "B_PAT2" and not row["ok"] and "forced" in row["error"]
        monkeypatch.undo()
        # the failed job is retried on the next apply
        assert [r.job_key for r in sync.apply(base_store).applied] == ["B_PAT2"]

    def test_apply_unknown_key(self, base_store: Store) -> None:
        with pytest.raises(sync.SyncError):
            sync.apply(base_store, ["NOPE"])


# ── edges ─────────────────────────────────────────────────────


class TestEdges:
    def test_maybe_instant_fallbacks(self) -> None:
        assert sync._maybe_instant("") is None and sync._maybe_instant(None) is None
        assert sync._maybe_instant("garbage") is None
        naive = sync._maybe_instant("2026-10-01 12:00:00")
        assert naive is not None and naive.tzinfo is timezone.utc
        aware = sync._maybe_instant(datetime(2026, 10, 1, tzinfo=timezone.utc))
        assert aware is not None and aware.hour == 0
        assert sync._maybe_instant(datetime(2026, 10, 1)).tzinfo is timezone.utc

    def test_incomplete_jobs_are_not_complete(self, tmp_path: Path) -> None:
        store = Store(path=str(tmp_path))
        store._append_job(
            key="NOFILES", endpoint="Group/all", types="Patient", status="complete"
        )
        store._append_job(
            key="HALF", endpoint="Group/all", types="Patient", status="complete"
        )
        store._append_file(
            job_key="HALF", resource_type="Patient", url="u1", status="complete"
        )
        store._append_file(
            job_key="HALF", resource_type="Patient", url="u2", status="failed"
        )
        assert sync.complete_jobs(store) == []
        assert sync.status(store)["jobs"] == []

    def test_empty_delta_counts_when_manifest_says_so(self, base_store: Store) -> None:
        """A delta with nothing to deliver has no files; its manifest makes it a job and moves the watermark."""
        sync.apply(base_store)
        base_store._append_job(
            key="D_EMPTY",
            endpoint="Group/all",
            types="Patient",
            since=sync.format_instant(T0),
            status="complete",
            transaction_time=sync.format_instant(T0 + timedelta(days=1)),
        )
        assert "D_EMPTY" not in {j.key for j in sync.complete_jobs(base_store)}
        base_store._write_manifest(
            "D_EMPTY",
            {
                "transactionTime": sync.format_instant(T0 + timedelta(days=1)),
                "output": [],
            },
        )
        assert "D_EMPTY" in {j.key for j in sync.complete_jobs(base_store)}
        rep = sync.apply(base_store)
        assert rep.applied[0].job_key == "D_EMPTY" and rep.ok
        assert rep.applied[0].counts["patient"]["rows"] == 0
        assert rep.applied[0].counts["patient"]["current"] == 94
        assert sync.next_since(base_store)["Patient"] == sync.format_instant(
            T0 + timedelta(days=1)
        )

    def test_watermark_filters(self, base_store: Store) -> None:
        jobs = sync.complete_jobs(base_store)
        assert sync.watermarks(jobs, endpoint="Group/runout") == {}
        assert len(sync.watermarks(jobs, endpoint="Patient")) == 3  # same population
        assert sync.watermarks(jobs, type_filter="x") == {}
        assert len(sync.watermarks(jobs, endpoint="Group/all", type_filter="")) == 3
        assert set(sync.baselines(jobs)) == {
            ("current", "", t) for t in sync.RESOURCE_TYPES
        }

    def test_missing_export_dir(self, tmp_path: Path) -> None:
        store = Store(path=str(tmp_path))
        store._append_job(
            key="GONE", endpoint="Group/all", types="Patient", status="complete"
        )
        store._append_file(
            job_key="GONE", resource_type="Patient", url="u", status="complete"
        )
        with pytest.raises(sync.SyncError, match="no export directory"):
            sync.apply(store)
        row = sync.read_ledger(store)[-1]
        assert (
            row["job_key"] == "GONE" and not row["ok"] and "SyncError" in row["error"]
        )

    def test_delta_older_than_baseline_is_ignored(self, base_store: Store) -> None:
        sync.apply(base_store)
        delta, _ = _delta_rows(
            _rows("ExplanationOfBenefit"), _rows("Patient"), T0 - timedelta(days=10)
        )
        add_job(
            base_store,
            "D_OLD",
            {"ExplanationOfBenefit": delta},
            since=sync.format_instant(T0 - timedelta(days=10)),
            transaction_time=T0 - timedelta(days=3),
        )
        rep = sync.apply(base_store)
        c = rep.applied[0].counts["explanation_of_benefit"]
        assert c["stale_ignored"] == 10 and c["current"] == 269 and c["new"] == 0
        assert any("predates baseline" in n for n in rep.applied[0].notes)

    def test_job_claiming_a_type_it_did_not_deliver(self, base_store: Store) -> None:
        sync.apply(base_store)
        # the job row says Patient,Coverage but only a Coverage file landed
        d = Path(base_store.root) / "exports" / "D_MIX"
        d.mkdir()
        cov = copy.deepcopy(_rows("Coverage")[0])
        _lu(cov, T0 + timedelta(days=2))
        (d / "Coverage.ndjson").write_text(json.dumps(cov) + "\n")
        base_store._append_file(
            job_key="D_MIX",
            resource_type="Coverage",
            url="u",
            storage_path=str(d / "Coverage.ndjson"),
            status="complete",
        )
        base_store._append_job(
            key="D_MIX",
            endpoint="Group/all",
            types="Patient,Coverage",
            since=sync.format_instant(T0),
            status="complete",
            transaction_time=sync.format_instant(T0 + timedelta(days=2)),
        )
        rep = sync.apply(base_store)
        c = rep.applied[0].counts
        assert c["patient"]["rows"] == 0 and c["patient"]["current"] == 94
        assert c["coverage"]["replaced"] == 1 and c["coverage"]["current"] == 175

    def test_invariants_catch_duplicates_conservation_and_strays(
        self, base_store: Store
    ) -> None:
        import duckdb

        con = duckdb.connect()
        con.execute(
            "create table cur_patient as select * from (values ('A','A'), ('A','A')) t(bene_id, id)"
        )
        con.execute(
            "create table cur_explanation_of_benefit (clm_cntl_num varchar, clm_cntl_num_match varchar, "
            "clm_natural_key varchar, clm_orig_cntl_num varchar, superseded_by varchar, clm_id varchar, id varchar)"
        )
        con.execute("create table cur_eob_item as select 'X' as clm_id")
        job = sync.complete_jobs(base_store)[0]  # the Patient baseline
        rep = sync.JobReport(
            job_key=job.key,
            kind="baseline",
            endpoint="Group/all",
            type_filter="",
            types=["Patient"],
            since="",
            transaction_time="",
            counts={"patient": {**dict.fromkeys(sync.COUNT_FIELDS, 0), "current": 2}},
        )
        with pytest.raises(sync.SyncInvariantError) as exc:
            sync._invariants(con, base_store, job, rep, {"patient": 5}, {})
        msg = str(exc.value)
        assert "2 rows but 1 distinct keys" in msg
        assert "count conservation failed" in msg and "5 + 0 new" in msg
        assert "1 line(s) without a current header" in msg
        con.close()

    def test_older_baseline_note_in_invariants(self, base_store: Store) -> None:
        sync.apply(base_store)
        add_job(
            base_store,
            "B_PAT0",
            {"Patient": _rows("Patient")},
            transaction_time=T0 - timedelta(days=1),
        )
        rep = sync.apply(base_store)
        assert any(
            "newest baseline stays authoritative" in n for n in rep.applied[0].notes
        )


# ── upstream review gaps (#1293–#1298) ───────────────────────


def _manifest_with_errors(store: Store, key: str, tt: datetime, n: int = 1) -> None:
    """Mark *key* partial: manifest lists OperationOutcome files and the files ledger has them."""
    man = store.manifest(key)
    man["error"] = [
        {"type": "OperationOutcome", "url": f"https://s/data/{key}/e{i}-error.ndjson"}
        for i in range(n)
    ]
    store._write_manifest(key, man)
    d = Path(store.root) / "exports" / key
    for i in range(n):
        p = d / (
            "OperationOutcome.ndjson" if i == 0 else f"OperationOutcome.{i}.ndjson"
        )
        p.write_text(
            '{"resourceType":"OperationOutcome","issue":[{"severity":"error","code":"exception","diagnostics":"BFD timeout for bene"}]}\n'
        )
        store._append_file(
            job_key=key,
            resource_type="OperationOutcome",
            url=man["error"][i]["url"],
            storage_path=str(p),
            file_size=p.stat().st_size,
            status="complete",
        )


class TestPartialJobs:
    def test_partial_delta_applies_but_does_not_advance(
        self, base_store: Store
    ) -> None:
        sync.apply(base_store)
        since = T0 + timedelta(minutes=2)
        delta, _ = _delta_rows(_rows("ExplanationOfBenefit"), _rows("Patient"), since)
        add_job(
            base_store,
            "D_PART",
            {"ExplanationOfBenefit": delta},
            since=sync.format_instant(since),
            transaction_time=since + timedelta(days=7),
        )
        _manifest_with_errors(base_store, "D_PART", since + timedelta(days=7), n=2)
        info = next(j for j in sync.complete_jobs(base_store) if j.key == "D_PART")
        assert info.partial and info.error_files == 2
        # its rows still reconcile (they are real), but the watermark stays at the baseline
        rep = sync.apply(base_store)
        assert rep.ok and rep.applied[0].counts["explanation_of_benefit"]["new"] == 3
        assert sync.next_since(base_store)[
            "ExplanationOfBenefit"
        ] == sync.format_instant(since)
        checks = sync.check_since(
            base_store,
            sync.format_instant(since),
            types=["ExplanationOfBenefit"],
            now=since + timedelta(days=8),
        )
        codes = {c.code: c for c in checks}
        assert (
            "partial_job" in codes
            and "D_PART (2 file(s))" in codes["partial_job"].message
        )
        st = sync.status(base_store)
        row = next(j for j in st["jobs"] if j["key"] == "D_PART")
        assert row["partial"] and row["error_files"] == 2
        assert (
            base_store.error_diagnostics("D_PART")[0]["diagnostics"]
            == "BFD timeout for bene"
        )


class TestClaimIdentity:
    def _eob_with(
        self,
        base: dict,
        *,
        new_id: str,
        ccn: str | None,
        related_to: str | None,
        lu: datetime,
        system: str | None = None,
    ) -> dict:
        e = copy.deepcopy(base)
        e["id"] = new_id
        e["identifier"] = [i for i in e["identifier"] if not i.get("system") == CCN]
        if ccn:
            e["identifier"].append({"system": CCN, "value": ccn})
        if related_to:
            e["related"] = [{"reference": {"system": CCN, "value": related_to}}]
        if system:
            for t in e["meta"]["tag"]:
                if t.get("system", "").endswith("System-Type"):
                    t["code"] = system
        _lu(e, lu)
        return e

    def test_new_control_number_supersedes_the_old_claim(
        self, base_store: Store
    ) -> None:
        sync.apply(base_store)
        eobs = _rows("ExplanationOfBenefit")
        finals = [
            e
            for e in eobs
            if any(t.get("code") == "FinalAction" for t in e["meta"]["tag"])
        ]
        a = finals[0]
        a_ccn = _ccn(a)
        since = T0 + timedelta(minutes=2)
        b = self._eob_with(
            a,
            new_id=a["id"] + "-B",
            ccn=a_ccn + "B",
            related_to=a_ccn,
            lu=since + timedelta(hours=1),
        )
        c = self._eob_with(
            a,
            new_id=a["id"] + "-C",
            ccn=a_ccn + "C",
            related_to=a_ccn + "B",
            lu=since + timedelta(hours=2),
        )
        add_job(
            base_store,
            "D_SUP",
            {"ExplanationOfBenefit": [b, c]},
            since=sync.format_instant(since),
            transaction_time=since + timedelta(days=1),
        )
        rep = sync.apply(base_store)
        cnt = rep.applied[0].counts["explanation_of_benefit"]
        assert (
            cnt["new"] == 2 and cnt["adjusted"] == 2 and cnt["superseded"] == 1
        )  # B is superseded by C
        import duckdb

        flat = Path(base_store.root) / "flat"
        con = duckdb.connect()
        rows = con.execute(
            f"select id, superseded_by from read_parquet('{flat / 'explanation_of_benefit.parquet'}') "
            f"where clm_cntl_num in ('{a_ccn}', '{a_ccn}B', '{a_ccn}C') order by id"
        ).fetchall()
        by_id = dict(rows)
        assert by_id[a["id"]] == a_ccn + "B" and by_id[a["id"] + "-B"] == a_ccn + "C"
        assert by_id[a["id"] + "-C"] is None
        sup = con.execute(
            f"select count(*) from read_parquet('{flat / 'eob_supersession.parquet'}')"
        ).fetchone()[0]
        assert sup >= 3
        # the connector sees one live claim for the chain
        import polars as pl

        from bcda.express.cclf import stg_claim_header

        hdr = pl.read_parquet(flat / "explanation_of_benefit.parquet")
        xref = pl.DataFrame(
            {"bene_id": hdr["bene_id"].unique().to_list()}
        ).with_columns(pl.lit("M").alias("current_bene_mbi_id"))
        out = stg_claim_header(
            bcda__explanation_of_benefit=hdr, bcda___stg_beneficiary_xref=xref
        )
        live = out.filter(
            pl.col("clm_cntl_num").is_in([a_ccn, a_ccn + "B", a_ccn + "C"])
        )
        assert live["id"].to_list() == [a["id"] + "-C"]
        con.close()

    def test_dme_vms_digit_matches_the_nch_claim(self, base_store: Store) -> None:
        sync.apply(base_store)
        eobs = _rows("ExplanationOfBenefit")
        dme = next(
            e
            for e in eobs
            if any(
                c.get("code") == "DME" for c in (e.get("type") or {}).get("coding", [])
            )
            and any(t.get("code") == "FinalAction" for t in e["meta"]["tag"])
        )
        since = T0 + timedelta(minutes=2)
        vms = self._eob_with(
            dme,
            new_id=dme["id"] + "-VMS",
            ccn="1" + _ccn(dme),
            related_to=None,
            lu=since + timedelta(hours=1),
            system="SharedSystem",
        )
        add_job(
            base_store,
            "D_VMS",
            {"ExplanationOfBenefit": [vms]},
            since=sync.format_instant(since),
            transaction_time=since + timedelta(days=1),
            type_filter="ExplanationOfBenefit?_tag=https://bluebutton.cms.gov/fhir/CodeSystem/System-Type|SharedSystem",
        )
        rep = sync.apply(base_store)
        cnt = rep.applied[0].counts["explanation_of_benefit"]
        # same claim, newer version: replaced, not new
        assert cnt["new"] == 0 and cnt["replaced"] == 1 and cnt["current"] == 269
        import duckdb

        flat = Path(base_store.root) / "flat"
        con = duckdb.connect()
        row = con.execute(
            f"select id, clm_cntl_num, clm_cntl_num_match, system_type from read_parquet('{flat / 'explanation_of_benefit.parquet'}') "
            f"where clm_cntl_num_match = '{_ccn(dme)}'"
        ).fetchall()
        assert row == [(dme["id"] + "-VMS", "1" + _ccn(dme), _ccn(dme), "SharedSystem")]
        con.close()

    def test_pde_natural_key(self) -> None:
        from datetime import date

        from bcda.express import v3

        assert v3.pde_natural_key(
            "B1", date(2026, 1, 2), "01", "1234567890", "RX9", 2.0
        ) == ("pde:B1|2026-01-02|01|1234567890|RX9|2")
        assert v3.pde_natural_key("B1", None, None, None, None, None) is None
        assert (
            v3.pde_natural_key("B1", "2026-01-02", "", "", "RX9", 1.5)
            == "pde:B1|2026-01-02|||RX9|1.5"
        )
        assert (
            v3.match_control_number("1222222222222222", "SharedSystem", "DME")
            == "222222222222222"
        )
        assert (
            v3.match_control_number("1222222222222222", "NationalClaimsHistory", "DME")
            == "1222222222222222"
        )
        assert v3.match_control_number("1999", "SharedSystem", "CARRIER") == "1999"
        assert v3.match_control_number(None, "SharedSystem", "DME") is None

    def test_legacy_current_tables_without_identity_columns(
        self, base_store: Store
    ) -> None:
        """Current tables and partitions flattened before the identity columns existed still reconcile."""
        import duckdb

        sync.apply(base_store)
        flat = Path(base_store.root) / "flat"
        con = duckdb.connect()
        for p in [
            flat / "explanation_of_benefit.parquet",
            *flat.glob("history/*/explanation_of_benefit.parquet"),
        ]:
            con.execute(
                f"copy (select * exclude (clm_cntl_num_match, clm_natural_key, superseded_by, prvdr_srvc_id_qlfyr_cd) "
                f"from read_parquet('{p}')) to '{p}.tmp' (format parquet)"
            )
            Path(f"{p}.tmp").replace(p)
        con.close()
        since = T0 + timedelta(minutes=2)
        delta, expect = _delta_rows(
            _rows("ExplanationOfBenefit"), _rows("Patient"), since
        )
        add_job(
            base_store,
            "D_EOB",
            {"ExplanationOfBenefit": delta},
            since=sync.format_instant(since),
            transaction_time=since + timedelta(days=7),
        )
        rep = sync.apply(base_store)
        c = rep.applied[0].counts["explanation_of_benefit"]
        assert {k: c[k] for k in expect} == expect


class TestAttributionAndRunoutChecks:
    def test_attribution_expired_floor_and_runout(self, base_store: Store) -> None:
        sync.apply(base_store)
        now = T0 + timedelta(days=10)
        attribution = {
            "ingestion_dates": [
                {
                    "type": "last_attribution_update",
                    "timestamp": sync.format_instant(T0 - timedelta(days=30)),
                }
            ],
            "expiration_dates": [
                {
                    "type": "attribution_access_expiration",
                    "timestamp": sync.format_instant(now - timedelta(days=1)),
                }
            ],
        }
        checks = sync.check_since(
            base_store,
            "2026-10-01T12:00:00.000+00:00",
            now=now,
            attribution=attribution,
            types=["Patient"],
        )
        levels = {c.code: c.level for c in checks}
        assert levels["attribution_expired"] == "block"
        # floor: a job row recorded an attribution update newer than the proposed _since
        base_store._append_job(
            key="B_PAT",
            endpoint="Group/all",
            types="Patient",
            status="complete",
            transaction_time=sync.format_instant(T0),
            attribution_update=sync.format_instant(T0 - timedelta(days=5)),
        )
        checks = sync.check_since(
            base_store,
            sync.format_instant(T0 - timedelta(days=6)),
            now=now,
            types=["Patient"],
        )
        levels = {c.code: c.level for c in checks}
        assert levels["attribution_floor"] == "warn" and levels["overlap"] == "ok"
        # runout after July: warn (and endpoint_mix block because the store has no runout baseline)
        add_job(
            base_store,
            "R_PAT",
            {"Patient": _rows("Patient")[:3]},
            transaction_time=T0,
            endpoint="Group/runout",
        )
        checks = sync.check_since(
            base_store,
            sync.format_instant(T0),
            now=datetime(2026, 10, 25, tzinfo=timezone.utc),
            endpoint="Group/runout",
            types=["Patient"],
        )
        levels = {c.code: c.level for c in checks}
        assert levels["runout_window"] == "warn" and "endpoint_mix" not in levels
        checks = sync.check_since(
            base_store,
            sync.format_instant(T0),
            now=datetime(2026, 5, 1, tzinfo=timezone.utc),
            endpoint="Group/runout",
            types=["Patient"],
        )
        assert "runout_window" not in {c.code for c in checks}


class TestPartDNaturalKey:
    def test_pde_without_control_number_reconciles_by_natural_key(
        self, base_store: Store
    ) -> None:
        sync.apply(base_store)
        eobs = _rows("ExplanationOfBenefit")
        pde = next(
            e
            for e in eobs
            if any(
                c.get("code") == "PDE" for c in (e.get("type") or {}).get("coding", [])
            )
            and any(t.get("code") == "FinalAction" for t in e["meta"]["tag"])
        )
        since = T0 + timedelta(minutes=2)
        # first version: no control number at all → keyed by the natural key
        v1 = copy.deepcopy(pde)
        v1["id"] = pde["id"] + "-nk1"
        v1["identifier"] = [i for i in v1["identifier"] if i.get("system") != CCN]
        _lu(v1, since + timedelta(hours=1))
        v2 = copy.deepcopy(v1)
        v2["id"] = pde["id"] + "-nk2"
        _lu(v2, since + timedelta(hours=2))
        add_job(
            base_store,
            "D_PDE1",
            {"ExplanationOfBenefit": [v1]},
            since=sync.format_instant(since),
            transaction_time=since + timedelta(days=1),
        )
        add_job(
            base_store,
            "D_PDE2",
            {"ExplanationOfBenefit": [v2]},
            since=sync.format_instant(since + timedelta(days=1)),
            transaction_time=since + timedelta(days=2),
        )
        rep = sync.apply(base_store)
        c1 = rep.applied[0].counts["explanation_of_benefit"]
        c2 = rep.applied[1].counts["explanation_of_benefit"]
        assert c1["new"] == 1 and c2["new"] == 0 and c2["replaced"] == 1
        assert c2["current"] == 270
        import duckdb

        flat = Path(base_store.root) / "flat"
        con = duckdb.connect()
        rows = con.execute(
            f"select id, clm_natural_key from read_parquet('{flat / 'explanation_of_benefit.parquet'}') "
            "where clm_natural_key is not null"
        ).fetchall()
        assert rows == [(pde["id"] + "-nk2", rows[0][1])] and rows[0][1].startswith(
            "pde:"
        )
        con.close()


class TestMoreEdges:
    def test_type_claimed_but_never_delivered_anywhere(self, tmp_path: Path) -> None:
        """A job row naming a type no partition carries: counts recorded, nothing to reconcile."""
        store = Store(path=str(tmp_path / "s"))
        add_job(store, "B1", {"Patient": _rows("Patient")[:5]}, transaction_time=T0)
        # rewrite the job row to claim Coverage too
        store._append_job(
            key="B1",
            endpoint="Group/all",
            types="Patient,Coverage",
            status="complete",
            transaction_time=sync.format_instant(T0),
        )
        rep = sync.apply(store)
        c = rep.applied[0].counts
        assert (
            c["patient"]["current"] == 5
            and c["coverage"]["current"] == 0
            and c["coverage"]["rows"] == 0
        )

    def test_invariant_live_predecessor(self, base_store: Store) -> None:
        import duckdb

        con = duckdb.connect()
        con.execute(
            "create table cur_explanation_of_benefit (clm_cntl_num varchar, clm_cntl_num_match varchar, "
            "clm_natural_key varchar, clm_orig_cntl_num varchar, superseded_by varchar, clm_id varchar, id varchar)"
        )
        con.execute(
            "insert into cur_explanation_of_benefit values ('A','A',null,null,null,'a','a'), ('B','B',null,'A',null,'b','b')"
        )
        job = sync.complete_jobs(base_store)[0]
        rep = sync.JobReport(
            job_key=job.key,
            kind="baseline",
            endpoint="Group/all",
            type_filter="",
            types=["Patient"],
            since="",
            transaction_time="",
            counts={},
        )
        with pytest.raises(sync.SyncInvariantError, match="still named as predecessor"):
            sync._invariants(con, base_store, job, rep, {}, {})
        con.close()