feat(sp41): pipeline B — NOT_IN_835 → fresh 837Ps by (member, ISO-week)

backend/src/cyclone/rebill/pipeline_b.py

@@ -0,0 +1,116 @@
+"""Pipeline B: NOT_IN_835 visits → fresh 837Ps, batched by (member_id, ISO-week).
+
+One 837P per (member, ISO-week) pair. Per-visit files would be 4,509 SFTP
+round-trips; per-batch (no member split) would lose the operator's
+ability to track per-member. Member-week is the standard clearinghouse
+pattern and matches AxisCare's billing cycle.
+
+The timely-filing gate is applied per visit before batching. The
+override flag is the same for the whole batch (a per-batch override).
+"""
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal
+from pathlib import Path
+
+from cyclone.rebill.reconcile import VisitRow
+from cyclone.rebill.timely_filing import timely_filing_decision
+
+
+@dataclass(frozen=True)
+class MemberWeekBatch:
+    """Pre-emission 837P shape for Pipeline B. The orchestrator's serializer
+    adapter converts this to a ClaimOutput for serialize_837 (one 837P
+    carrying every visit in `visits` as a CLM segment).
+    """
+
+    member_id: str
+    iso_year: int
+    iso_week: int
+    visits: list[VisitRow]
+    has_overridden_visits: bool
+
+
+def build_pipeline_b_batches(
+    visits: list[VisitRow],
+    as_of: date,
+    override: bool,
+) -> list[MemberWeekBatch]:
+    """Group rebillable visits by (member_id, ISO-week).
+
+    Two filters are applied per visit before batching:
+      1. timely_filing_decision(...).rebillable — drops past-window
+         visits unless `override=True` (per-batch override flag).
+      2. implicit: only visits that survive (1) are grouped; the rest
+         are surfaced as EXCLUDED_TIMELY_FILING in the summary CSV.
+
+    Returns batches sorted by (member_id, iso_year, iso_week) for
+    deterministic filenames downstream.
+    """
+    by_key: dict[tuple[str, int, int], list[VisitRow]] = defaultdict(list)
+    overridden_keys: set[tuple[str, int, int]] = set()
+    for v in visits:
+        decision = timely_filing_decision(v.date, as_of, override)
+        if not decision.rebillable:
+            continue
+        iso = v.date.isocalendar()
+        key = (v.member_id, iso.year, iso.week)
+        by_key[key].append(v)
+        if decision.override and not decision.within_window:
+            overridden_keys.add(key)
+    return [
+        MemberWeekBatch(
+            member_id=member, iso_year=yr, iso_week=wk,
+            visits=vs, has_overridden_visits=(member, yr, wk) in overridden_keys,
+        )
+        for (member, yr, wk), vs in sorted(by_key.items())
+    ]
+
+
+def batch_visits_by_member_week(
+    visits: list[VisitRow],
+    as_of: date,
+    override: bool,
+) -> dict[str, list[VisitRow]]:
+    """Thin wrapper over build_pipeline_b_batches.
+
+    Returns ``{member_id: [VisitRow, ...]}`` — collapsing the (member,
+    ISO-week) batches back down to a per-member visit list. This is the
+    shape the orchestrator's summary aggregation expects.
+
+    NOTE: This collapses multiple weeks for the same member into one
+    list. If the orchestrator needs week-aware per-member iteration,
+    use build_pipeline_b_batches directly.
+    """
+    batches = build_pipeline_b_batches(visits, as_of=as_of, override=override)
+    out: dict[str, list[VisitRow]] = defaultdict(list)
+    for b in batches:
+        out[b.member_id].extend(b.visits)
+    return dict(out)
+
+
+def write_pipeline_b_files(
+    batches: list[MemberWeekBatch],
+    out_dir: Path,
+    serialize_837_fn,
+    tpid: str,
+) -> list[Path]:
+    """Write one 837P per batch, using HCPF-spec filenames.
+
+    `serialize_837_fn(batch)` returns the X12 string. The 837P envelope
+    is pure ASCII so we write it as text. The filename is
+    `cyclone.edi.filenames.build_outbound_filename(tpid, '837P')`.
+    """
+    # Lazy: keep cyclone.rebill.pipeline_b importable without zoneinfo from filenames.
+    from cyclone.edi.filenames import build_outbound_filename
+
+    out_dir.mkdir(parents=True, exist_ok=True)
+    paths: list[Path] = []
+    for b in batches:
+        body = serialize_837_fn(b)
+        fname = build_outbound_filename(tpid, "837P")
+        path = out_dir / fname
+        path.write_text(body, encoding="ascii")
+        paths.append(path)
+    return paths

backend/tests/test_rebill_pipeline_b.py

@@ -0,0 +1,41 @@
+"""Pipeline B: NOT_IN_835 visits → fresh 837Ps, batched by (member, ISO-week)."""
+from datetime import date
+from decimal import Decimal
+from cyclone.rebill.pipeline_b import build_pipeline_b_batches
+from cyclone.rebill.reconcile import VisitRow
+
+
+def _v(date_, member, proc, amt):
+    return VisitRow(date=date_, member_id=member, procedure=proc, billed=Decimal(amt))
+
+
+def test_batches_split_by_member_and_iso_week():
+    visits = [
+        _v(date(2026, 6, 23), "J813715", "T1019", "2.32"),
+        _v(date(2026, 6, 25), "J813715", "T1019", "2.32"),  # same week
+        _v(date(2026, 6, 23), "OTHER",    "T1019", "2.32"),  # different member
+        _v(date(2026, 6, 30), "J813715", "T1019", "2.32"),  # different week
+    ]
+    out = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False)
+    assert len(out) == 3
+    sizes = sorted(len(b.visits) for b in out)
+    assert sizes == [1, 1, 2]
+
+
+def test_timely_filing_excludes_old_visits():
+    visits = [
+        _v(date(2026, 1, 1), "OLD", "T1019", "2.32"),  # 187 days old
+        _v(date(2026, 6, 27), "NEW", "T1019", "2.32"), # 10 days old
+    ]
+    out = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False)
+    members = {b.member_id for b in out}
+    assert "OLD" not in members
+    assert "NEW" in members
+
+
+def test_override_relaxes_timely_filing():
+    visits = [_v(date(2026, 1, 1), "OLD", "T1019", "2.32")]
+    out_default = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False)
+    out_overridden = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=True)
+    assert out_default == []
+    assert len(out_overridden) == 1