feat(spot-check): delegate to parse_835_svc + reconcile with member-scoped CAS

Per the SP41-goal verifier feedback on AC2/AC3: the committed
select_spot_check_visits reimplemented reconcile inline against
service_line_payments on (procedure, DOS) only. That table has
no member_id at SVC scope, so OA-18 dedup inherited CAS reasons
from all SVCs at the same (procedure, DOS), misclassifying visits
across members (the operator's 'multiple rejections due to files
being submitted multiple times' noise).

This commit restructures the spot-check pipeline per the
strategist recommendation:

1. Extract load_in_window_svc_rows(ingest_dir, window_start,
   window_end) in parse_835_svc.py as the canonical ingest helper.
   run.py's existing loop is replaced with a one-line call.

2. select_spot_check_visits becomes a thin adapter:
   - Reads in-window visits from the visits table.
   - Loads SVCs via load_in_window_svc_rows.
   - Calls reconcile_visits_to_835 (authoritative match key).
   - Builds a member-scoped CAS lookup keyed by
     (member_id, procedure, svc_date) so OA-18 inspection only
     sees the matched member's adjudications.
   - Returns the top-N DENIED / PARTIAL / NOT_IN_835 by
     billed_amount desc.

3. New test_select_spot_check_visits_member_isolation unit test
   drives the SHIPPED code on the real path (real .x12 fixtures
   reparsed by parse_835_svc) and pins the structural fix:
   two visits at the same (procedure, DOS) for DIFFERENT members,
   member A with OA-18-only CAS, member B with none. A is
   excluded, B stays a candidate. The OLD (procedure, DOS)-only
   indexer excluded BOTH (verified separately at
   /tmp/old_logic_check.py).

4. Existing OA-18 test rewritten to use real .x12 fixtures in
   tmp_path (instead of inserting into service_line_payments /
   cas_adjustments, which the new pipeline does not read).

Verification:
- Full backend suite: 1593 passed, 10 skipped (prodfile/docker), 0 failed.
- Live Edifabric 10/10 re-run with the new pipeline: 10 files,
  all status='success', all 6 required segment classes per file.
  Evidence at /tmp/grok-goal-4336f8e1af6d/implementer/spot-check.json

backend/src/cyclone/rebill/parse_835_svc.py

@@ -4,7 +4,17 @@ Walks an 835 file segment-by-segment, propagating the CLP-scope NM1*QC
 NM109 (member_id) to every SVC row that follows. Captures DTM*472
 service dates and CAS adjustments (which may appear before or after
 DTM*472 in the segment stream).
+
+:func:`load_in_window_svc_rows` is the canonical ingest helper — it
+walks a directory of ``*.835`` / ``*.x12`` files (skipping AppleDouble
+shadow files), reparses each through :func:`parse_835_svc`, and
+returns only the rows whose ``svc_date`` falls inside the inclusive
+``[window_start, window_end]`` window. This is the single ingest path
+used by the SP41 rebill pipeline (see ``cyclone.rebill.run``) AND by
+the SP41-goal spot-check pipeline (see
+``cyclone.rebill.spot_check_pipeline``).
 """
+from collections.abc import Iterable, Iterator
 from dataclasses import dataclass
 from datetime import date
 from decimal import Decimal
@@ -121,4 +131,53 @@ def _walk_for_dtm_and_cas(
             # CAS*GR*CODE*AMT*QTY*AMT*QTY...  →  GR-CODE
             if len(ej) >= 3:
                 cas_reasons.append(f"{ej[1]}-{ej[2]}")
-    return svc_date, tuple(cas_reasons)
+    return svc_date, tuple(cas_reasons)
+
+
+def load_in_window_svc_rows(
+    ingest_dir: str | Path,
+    window_start: date,
+    window_end: date,
+    *,
+    file_glob: Iterable[str] = ("*.835", "*.x12"),
+) -> list[SvcRow]:
+    """Walk ``ingest_dir`` for 835 files, reparse, filter to DOS window.
+
+    The single canonical ingest path. Skips AppleDouble shadow files
+    (``._*`` — macOS resource forks surfaced alongside real files by
+    SFTP clients; the convention is used by ``cyclone.cli``,
+    ``cyclone.submission.core``, and ``cyclone.api_routers.submission``).
+
+    Args:
+        ingest_dir: Directory to walk. Both ``*.835`` and ``*.x12`` are
+            scanned (sorted by name for determinism).
+        window_start: Inclusive lower bound on ``SvcRow.svc_date``.
+        window_end: Inclusive upper bound on ``SvcRow.svc_date``.
+        file_glob: Optional override on the file extensions to walk;
+            defaults to the production pair.
+
+    Returns:
+        A flat list of :class:`SvcRow` for every SVC segment whose
+        ``svc_date`` falls inside the window. SVCs with no
+        ``DTM*472`` (and therefore ``svc_date is None``) are dropped
+        — we cannot bucket them into a window.
+
+    No reconciliation logic lives here — that is the responsibility of
+    :func:`cyclone.rebill.reconcile.reconcile_visits_to_835`. This
+    helper just yields the raw SVC rows indexed downstream.
+    """
+    ingest_dir_p = Path(ingest_dir)
+    files: list[Path] = []
+    for pattern in file_glob:
+        files.extend(ingest_dir_p.glob(pattern))
+    files = sorted(set(files))
+    svcs: list[SvcRow] = []
+    for p in files:
+        if p.name.startswith("._"):
+            continue
+        for s in parse_835_svc(p):
+            if s.svc_date is None:
+                continue
+            if window_start <= s.svc_date <= window_end:
+                svcs.append(s)
+    return svcs

backend/src/cyclone/rebill/run.py

@@ -18,7 +18,7 @@ from pathlib import Path
 from cyclone import edifabric as _edifabric
 from cyclone.edi.filenames import build_outbound_filename
 from cyclone.rebill.carc_filter import CarcDecision, decide_carc
-from cyclone.rebill.parse_835_svc import SvcRow, parse_835_svc
+from cyclone.rebill.parse_835_svc import SvcRow, load_in_window_svc_rows
 from cyclone.rebill.pipeline_a import RebillClaim
 from cyclone.rebill.pipeline_b import build_pipeline_b_batches
 from cyclone.rebill.reconcile import (
@@ -104,20 +104,9 @@ def run_rebill(
     visits_csv_p = Path(visits_csv_path or "data/source/apr-jun27.csv")
     ingest_dir_p = Path(ingest_dir or "ingest")
 
-    # 1) Ingest 835s — walk *.835 and *.x12, skip AppleDouble shadow files
+    # 1) Ingest 835s — canonical path through parse_835_svc.load_in_window_svc_rows
-    #    (._foo.835 are macOS resource forks that the SFTP client surfaces
+    #    (skips AppleDouble shadow files, filters to the DOS window).
-    #    alongside real files; matches the convention used by cli, submit-batch,
+    svcs = load_in_window_svc_rows(ingest_dir_p, window_start, window_end)
-    #    and api_routers/submission.py).
-    svcs: list[SvcRow] = []
-    for p in sorted(
-        list(ingest_dir_p.glob("*.835")) + list(ingest_dir_p.glob("*.x12"))
-    ):
-        if p.name.startswith("._"):
-            continue
-        for s in parse_835_svc(p):
-            # Guard against None svc_date (some 835s lack DTM*472 segments)
-            if s.svc_date is not None and window_start <= s.svc_date <= window_end:
-                svcs.append(s)
 
     # 2) Load visits CSV — header:
     #    ``Visit Date,Member ID,Procedure Code,Billable Amount``

backend/src/cyclone/rebill/spot_check_pipeline.py

@@ -2,29 +2,47 @@
 ``cyclone.rebill``).
 
 Splits the SP41 spot-check flow into a pure-Python generation step
-that consumes the canonical ``visits`` + ``service_line_payments``
+that consumes the canonical ``visits`` table and the in-window 835
-+ ``cas_adjustments`` tables and writes well-formed 837P files to
+SVC rows (reparsed from ``ingest/*.x12``) and writes well-formed
-disk. The companion validation step lives in
+837P files to disk. The companion validation step lives in
 :mod:`cyclone.rebill.spot_check_validate` (no fallback path —
 quota / transport errors fail loud).
 
 Public surface
 --------------
 
-- :func:`select_spot_check_visits` — DB-only: read the in-window
+- :func:`select_spot_check_visits` — read the in-window visits from
-  visits, reconcile against the ingested 835 SVC rows, drop
+  the ``visits`` table, reconcile against the 835 SVC rows reparsed
-  PAID / OA-18-only adjudications, return the top-N candidates by
+  from ``ingest_dir`` on the authoritative
-  ``billed_amount`` desc.
+  ``(member_id, procedure_code, service_date)`` key via
+  :func:`cyclone.rebill.reconcile.reconcile_visits_to_835`, drop
+  PAID / OA-18-only adjudications (member-scoped CAS inspection),
+  return the top-N candidates by ``billed_amount`` desc.
 - :func:`write_spot_check_files` — produce one well-formed 837P per
-  visit via the canonical ``cyclone.parsers.serialize_837.serialize_837``
+  visit via the canonical
-  single-claim path. Files are written segment-per-line to ``out_dir``
+  :func:`cyclone.parsers.serialize_837.serialize_837` single-claim
-  with the HCPF-spec filename via
+  path. Files are written segment-per-line to ``out_dir`` with the
+  HCPF-spec filename via
   :func:`cyclone.edi.filenames.build_outbound_filename`.
 
 The dedup of OA-18 (Exact Duplicate) adjudications reflects the
 operator's note that the 835s show multiple rejections because
 files were submitted multiple times; visits whose only adjudication
 is OA-18 on a duplicate SVC are NOT in-window rebill candidates.
+
+Structural invariant
+--------------------
+
+``select_spot_check_visits`` is a thin adapter — it does NOT
+re-implement reconciliation. The match key is owned by
+:func:`cyclone.rebill.reconcile.reconcile_visits_to_835`, which
+indexes SVCs on ``(member_id, procedure_code, service_date)`` — a
+key the ``service_line_payments`` table does not carry (no
+``member_id`` at SVC scope there). The previous version of this
+module re-implemented the join inline against ``service_line_payments``
+on ``(procedure, DOS)`` only, which let CAS reasons from one member
+bleed into another member's adjudication and misclassify visits as
+DENIED_DUPLICATE_NOISE.
 """
 from __future__ import annotations
 
@@ -37,10 +55,16 @@ from pathlib import Path
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from cyclone import db as db_mod
 from cyclone.db import Visit
 from cyclone.edi.filenames import build_outbound_filename
 from cyclone.parsers.serialize_837 import serialize_837
+from cyclone.rebill.parse_835_svc import SvcRow, load_in_window_svc_rows
+from cyclone.rebill.reconcile import (
+    OutcomeCategory,
+    ReconcileOutcome,
+    VisitRow as ReconcileVisitRow,
+    reconcile_visits_to_835,
+)
 from cyclone.rebill.spot_check import (
     VisitRow,
     build_claim_output,
@@ -59,6 +83,13 @@ SUBMITTER_CONTACT_PHONE = "8005550100"
 RECEIVER_NAME = "COLORADO MEDICAL ASSISTANCE PROGRAM"
 SBR09 = "MC"  # Medicaid
 
+# Single source of truth for the "this visit's only adjudication was
+# an OA-18 Exact Duplicate" classification. Used to drop visits
+# whose only adjudication is OA-18 (the operator's note about
+# prior duplicate submissions — those are NOT in-window rebill
+# candidates).
+DUPLICATE_NOISE_CARC = "OA-18"
+
 
 @dataclass(frozen=True)
 class SpotCheckFile:
@@ -70,7 +101,7 @@ class SpotCheckFile:
     claim_id: str
 
 
-def _visit_row_from_db(row: Visit) -> VisitRow:
+def _spot_check_visit_row_from_db(row: Visit) -> VisitRow:
     """Translate a ``Visit`` ORM row to a flat ``VisitRow``.
 
     The visits table stores modifiers colon-joined (post-normalize);
@@ -92,6 +123,22 @@ def _visit_row_from_db(row: Visit) -> VisitRow:
     )
 
 
+def _reconcile_visit_row_from_db(row: Visit) -> ReconcileVisitRow:
+    """Translate a ``Visit`` ORM row to ``reconcile.VisitRow`` (slim).
+
+    The reconcile module's ``VisitRow`` only carries the four fields
+    that drive the match key (``(member_id, procedure, date)`` plus
+    ``billed``); it deliberately drops the spot-check-only fields
+    (modifiers, icd10, prior_auth, client_name).
+    """
+    return ReconcileVisitRow(
+        date=row.dos,
+        member_id=row.member_id,
+        procedure=row.procedure_code,
+        billed=Decimal(str(row.billed_amount or 0)),
+    )
+
+
 def _build_claim_id(visit: VisitRow, idx: int) -> str:
     """Spot-check CLM01 — uppercase ``R`` prefix satisfies the plan's
     literal grep ``^CLM\\*[A-Z]`` (the lowercase ``r`` would not match).
@@ -100,18 +147,71 @@ def _build_claim_id(visit: VisitRow, idx: int) -> str:
     return f"R{visit.dos.isoformat().replace('-', '')}-{visit.member_id}-{idx:02d}"
 
 
+def _member_scoped_cas_lookup(
+    svc_rows: list[SvcRow],
+) -> dict[tuple[str, str, date], list[str]]:
+    """Index SVC CAS reasons by ``(member_id, procedure, svc_date)``.
+
+    Member-scoped so a DENIED visit's OA-18 inspection only sees the
+    adjudicated-by-this-member SVCs — not adjudications on the same
+    ``(procedure, DOS)`` for a different member.
+    """
+    by_key: dict[tuple[str, str, date], list[str]] = {}
+    for s in svc_rows:
+        if s.svc_date is None:
+            continue
+        if not s.cas_reasons:
+            continue
+        by_key.setdefault(
+            (s.member_id, s.procedure, s.svc_date), [],
+        ).extend(s.cas_reasons)
+    return by_key
+
+
+def _classify_denied(
+    matched_cas: list[str],
+) -> str:
+    """``"DENIED_DUPLICATE_NOISE"`` iff every matched CAS is ``OA-18``.
+
+    The bucket contains ONLY this visit's member-scoped SVC CAS
+    reasons, so OA-18 from a different member cannot bleed in.
+    Returns ``"DENIED"`` for any other CAS mix (including empty —
+    a DENIED with no CAS at all is a legitimate "adjudicated as
+    denied" outcome, not duplicate noise).
+    """
+    if matched_cas and set(matched_cas) == {DUPLICATE_NOISE_CARC}:
+        return "DENIED_DUPLICATE_NOISE"
+    return "DENIED"
+
+
 def select_spot_check_visits(
     session: Session,
     *,
     window_start: date,
     window_end: date,
     n: int = 10,
+    ingest_dir: str | Path | None = None,
 ) -> list[tuple[VisitRow, str, list[str]]]:
     """Pick ``n`` spot-check candidates from the canonical visits table.
 
-    Reads ``visits`` for the DOS window and ``service_line_payments`` /
+    Reads in-window visits from ``visits`` and reparses the in-window
-    ``cas_adjustments`` for the same window. Reconciles each visit on
+    835 SVC rows from ``ingest_dir`` via the canonical
-    ``(member_id, procedure_code, DOS)`` against the SVC rows.
+    :func:`cyclone.rebill.parse_835_svc.load_in_window_svc_rows`
+    helper. Reconciles each visit on
+    ``(member_id, procedure_code, DOS)`` through the proven
+    :func:`cyclone.rebill.reconcile.reconcile_visits_to_835`.
+
+    Args:
+        session: Open SQLAlchemy session on the canonical visits DB.
+        window_start: Inclusive DOS lower bound.
+        window_end: Inclusive DOS upper bound.
+        n: Top-N by ``billed_amount`` desc.
+        ingest_dir: Path to the 835 ingest directory (``.835`` / ``.x12``
+            files). Required: the visits table holds the canonical
+            member-of-record but not the adjudication; without a
+            reparse of the raw 835s we cannot classify anything
+            beyond NOT_IN_835. Defaults to ``"ingest"`` (the project
+            root path the production CLI uses).
 
     Returns:
         A list of ``(VisitRow, disposition, cas_reasons)`` tuples, top
@@ -122,92 +222,48 @@ def select_spot_check_visits(
     Disposition classification:
       * ``PAID`` — at least one matching SVC paid ≥ 95% of billed.
       * ``PARTIAL`` — at least one SVC paid but < 95%.
-      * ``DENIED`` — matching SVCs but no payment.
+      * ``DENIED`` — matching SVCs but no payment, with member-scoped
-      * ``DENIED_DUPLICATE_NOISE`` — pure OA-18 (Exact Duplicate) CAS.
+        CAS reasons NOT all ``OA-18``.
-        Excluded per the operator's note about prior duplicate
+      * ``DENIED_DUPLICATE_NOISE`` — pure ``OA-18`` (Exact Duplicate)
-        submissions — these are NOT in-window rebill candidates.
+        CAS for THIS member. Excluded per the operator's note about
+        prior duplicate submissions — these are NOT in-window rebill
+        candidates.
       * ``NOT_IN_835`` — no matching SVC at all.
     """
-    visits = session.execute(
+    db_visits = session.execute(
         select(Visit).where(
             Visit.dos >= window_start, Visit.dos <= window_end,
         )
     ).scalars().all()
 
-    svc_rows = session.execute(
+    # Build parallel lists so the outcome index aligns with the
-        db_mod.text(
+    # visit index — ``reconcile_visits_to_835`` preserves order.
-            """
+    spot_rows: list[VisitRow] = [_spot_check_visit_row_from_db(v) for v in db_visits]
-            SELECT slp.procedure_code AS procedure,
+    recon_rows: list[ReconcileVisitRow] = [_reconcile_visit_row_from_db(v) for v in db_visits]
-                   slp.service_date   AS svc_date,
-                   slp.charge         AS charge,
-                   slp.payment        AS paid,
-                   slp.remittance_id  AS remittance_id
-              FROM service_line_payments slp
-              JOIN remittances r ON r.id = slp.remittance_id
-             WHERE slp.service_date BETWEEN :start AND :end
-               AND r.is_reversal = 0
-            """
-        ),
-        {"start": window_start.isoformat(), "end": window_end.isoformat()},
-    ).mappings().all()
 
-    cas_by_remit: dict[str, list[str]] = {}
+    ingest_dir_p = Path(ingest_dir) if ingest_dir is not None else Path("ingest")
-    cas_rows = session.execute(
+    svc_rows = load_in_window_svc_rows(ingest_dir_p, window_start, window_end)
-        db_mod.text(
+    cas_lookup = _member_scoped_cas_lookup(svc_rows)
-            """
-            SELECT remittance_id, group_code, reason_code
-              FROM cas_adjustments
-             WHERE remittance_id IN (
-                 SELECT id FROM remittances WHERE is_reversal = 0
-             )
-            """
-        )
-    ).all()
-    for cr in cas_rows:
-        cas_by_remit.setdefault(str(cr.remittance_id), []).append(
-            f"{cr.group_code}-{cr.reason_code}"
-        )
 
-    # SQLite returns DATE columns as ISO strings; normalize to date.
+    outcomes: list[ReconcileOutcome] = reconcile_visits_to_835(recon_rows, svc_rows)
-    def _to_date(v) -> date:
-        if isinstance(v, date):
-            return v
-        return date.fromisoformat(str(v)[:10])
-
-    # Best-of-N SVC match by (procedure, DOS). The DB-side 835 parser
-    # (SP20) does not capture member_id at SVC scope, so we use the
-    # visits table's member_id directly — the visits table is the
-    # canonical member-of-record source, and an SVC at (procedure,
-    # DOS) for the same billed amount within 5% tolerance is
-    # unambiguous in practice.
-    by_pdos: dict[tuple[str, date], list] = {}
-    for s in svc_rows:
-        d = _to_date(s.svc_date)
-        by_pdos.setdefault((s.procedure, d), []).append(s)
 
     out: list[tuple[VisitRow, str, list[str]]] = []
-    for v in visits:
+    for vr, outcome in zip(spot_rows, outcomes):
-        vr = _visit_row_from_db(v)
+        if outcome.category == OutcomeCategory.NOT_IN_835:
-        cands = by_pdos.get((vr.procedure_code, vr.dos), [])
-        if not cands:
             out.append((vr, "NOT_IN_835", []))
             continue
-        total_paid = sum(
+        # DENIED / PAID / PARTIAL all had a matched SVC. CAS reasons
-            (Decimal(str(c.paid or 0)) for c in cands), Decimal("0")
+        # are pulled from the matched member's SVC row only — never
+        # from another member's adjudication at the same (procedure, DOS).
+        matched_cas = cas_lookup.get(
+            (vr.member_id, vr.procedure_code, vr.dos), [],
         )
-        any_paid = any(Decimal(str(c.paid or 0)) > 0 for c in cands)
+        if outcome.category == OutcomeCategory.PAID:
-        cas_reasons: list[str] = []
+            out.append((vr, "PAID", matched_cas))
-        for c in cands:
+        elif outcome.category == OutcomeCategory.PARTIAL:
-            cas_reasons.extend(cas_by_remit.get(str(c.remittance_id), []))
+            out.append((vr, "PARTIAL", matched_cas))
-        if any_paid and total_paid >= vr.billed_amount * Decimal("0.95"):
-            out.append((vr, "PAID", cas_reasons))
-        elif any_paid:
-            out.append((vr, "PARTIAL", cas_reasons))
         else:
-            if cas_reasons and set(cas_reasons) == {"OA-18"}:
+            out.append((vr, _classify_denied(matched_cas), matched_cas))
-                out.append((vr, "DENIED_DUPLICATE_NOISE", cas_reasons))
-            else:
-                out.append((vr, "DENIED", cas_reasons))
 
     # Keep DENIED / PARTIAL / NOT_IN_835; drop PAID + duplicate-noise.
     candidates = [(v, d, c) for (v, d, c) in out

backend/tests/test_spot_check_pipeline.py

@@ -11,6 +11,13 @@ HTTP seam) and unit-testable end-to-end:
 These tests pin the live-shape contract that the production driver
 exercises, so a regression in either module shows up here without
 having to run the live driver.
+
+The 835 ingestion path under test reads raw ``*.x12`` files via
+``cyclone.rebill.parse_835_svc.load_in_window_svc_rows`` (NOT the
+DB's ``service_line_payments`` / ``cas_adjustments`` tables, which
+have no ``member_id`` at SVC scope). Tests that need SVC rows
+write minimal 835 fixtures into ``tmp_path`` and pass
+``ingest_dir=tmp_path``.
 """
 from __future__ import annotations
 
@@ -24,7 +31,6 @@ from pathlib import Path
 
 import httpx
 import pytest
-from sqlalchemy import text
 
 import cyclone.edifabric as edifabric
 import cyclone.secrets as _secrets
@@ -51,7 +57,85 @@ _X12_INTERCHANGE = {
 }
 
 
-# --- Fixtures --------------------------------------------------------- #
+# --- 835 fixture builder --------------------------------------------- #
+
+
+def _build_835_with_svc(
+    *,
+    member_id: str,
+    claim_id: str,
+    status: str,
+    procedure: str,
+    svc_date: date,
+    charge: Decimal | str,
+    paid: Decimal | str,
+    cas: list[str] | None = None,
+    pay_date: date | None = None,
+) -> str:
+    """Build a minimal valid 835 interchange with one SVC.
+
+    The shape matches what ``cyclone.rebill.parse_835_svc.parse_835_svc``
+    walks: ISA/GS/ST header, BPR + TRN + DTM*405 (pay date), N1*PR + N1*PE
+    (payer / payee), LX (loop), CLP (claim), NM1*QC (member), SVC + DTM*472
+    (service date) + CAS (optional), SE/GE/IEA trailer.
+
+    ``cas`` is a list of ``"GR-CODE"`` (e.g. ``["OA-18"]``).
+    """
+    charge_s = str(Decimal(str(charge)))
+    paid_s = str(Decimal(str(paid)))
+    pay_dtm = (
+        f"DTM*405*{pay_date.strftime('%Y%m%d')}~"
+        if pay_date is not None
+        else ""
+    )
+    cas_segment = ""
+    if cas:
+        cas_segment = "~".join(
+            f"CAS*{c.replace('-', '*')}*0" for c in cas
+        ) + "~"
+    # Procedure in SVC: "HC:T1019:U2" format that parse_835_svc splits on ":"
+    svc_comp = f"HC:{procedure}"
+    segs = [
+        "ISA*00*          *00*          *ZZ*COMEDASSISTPROG*ZZ*11525703"
+        f"       *260202*0218*^*00501*200005613*0*P*:~",
+        "GS*HP*COMEDASSISTPROG*11525703*20260202*021851*200005613*X*005010X221A1~",
+        "ST*835*1001~",
+        "BPR*I*0*C*ACH*CCP*01*075911603*DA*00000004556640373*1811725341"
+        "**01*102103407*DA*8911987439*20260202~",
+        "TRN*1*003049968*1811725341~",
+        "REF*EV*1881068062~",
+        pay_dtm,
+        "N1*PR*CO_TXIX*XV*7912900843~",
+        "N3*P.O. BOX 30~",
+        "N4*DENVER*CO*80201~",
+        "PER*CX*MEDICAID PROVIDER SERVICES*TE*8442352387~",
+        "N1*PE*TOC, INC*XX*1881068062~",
+        "REF*TJ*721587149~",
+        "LX*1~",
+        f"CLP*{claim_id}*{status}*{charge_s}*{paid_s}**MC*1234~",
+        f"NM1*QC*1*Member*First****MR*{member_id}~",
+        f"SVC*{svc_comp}*{charge_s}*{paid_s}**1~",
+        f"DTM*472*{svc_date.strftime('%Y%m%d')}~",
+        cas_segment,
+        "SE*15*1001~",
+        "GE*1*200005613~",
+        "IEA*1*200005613~",
+    ]
+    return "".join(segs)
+
+
+def _write_835(
+    tmp_path: Path,
+    name: str,
+    content: str,
+) -> Path:
+    """Write a 835 fixture to ``tmp_path/<name>`` and return its path."""
+    p = tmp_path / name
+    p.write_text(content, encoding="ascii")
+    return p
+
+
+# --- Visits-CSV fixture ---------------------------------------------- #
 
 
 @pytest.fixture
@@ -129,13 +213,14 @@ def _install(handler):
 # --- Pipeline tests --------------------------------------------------- #
 
 
-def test_select_spot_check_visits_returns_in_window_only(loaded_db):
+def test_select_spot_check_visits_returns_in_window_only(loaded_db, tmp_path):
     """select_spot_check_visits honors the DOS window."""
     out = select_spot_check_visits(
         loaded_db,
         window_start=date(2026, 1, 1),
         window_end=date(2026, 6, 27),
         n=10,
+        ingest_dir=tmp_path,
     )
     assert len(out) == 2
     # The stale 2025 row was filtered out at the load step.
@@ -143,93 +228,179 @@ def test_select_spot_check_visits_returns_in_window_only(loaded_db):
     assert all(date(2026, 1, 1) <= d <= date(2026, 6, 27) for d in dos_list)
 
 
-def test_select_spot_check_visits_marks_not_in_835(loaded_db):
+def test_select_spot_check_visits_marks_not_in_835(loaded_db, tmp_path):
-    """With no SVC rows in DB, visits are NOT_IN_835 (rebill candidates)."""
+    """With no SVC rows in ingest/, visits are NOT_IN_835 (rebill candidates)."""
     out = select_spot_check_visits(
         loaded_db,
         window_start=date(2026, 1, 1),
         window_end=date(2026, 6, 27),
         n=10,
+        ingest_dir=tmp_path,
     )
     dispositions = [d for (_v, d, _c) in out]
     assert dispositions == ["NOT_IN_835", "NOT_IN_835"]
 
 
-def test_select_spot_check_visits_n_truncates(loaded_db):
+def test_select_spot_check_visits_n_truncates(loaded_db, tmp_path):
     """n=1 returns the top 1 by billed_amount desc."""
     out = select_spot_check_visits(
         loaded_db,
         window_start=date(2026, 1, 1),
         window_end=date(2026, 6, 27),
         n=1,
+        ingest_dir=tmp_path,
     )
     assert len(out) == 1
     assert out[0][0].member_id == "R111111"  # $300 > $250
 
 
-def test_select_spot_check_visits_oa18_excluded(loaded_db):
+def test_select_spot_check_visits_oa18_excluded_member_scoped(
-    """OA-18-only SVC matches are excluded as DENIED_DUPLICATE_NOISE."""
+    loaded_db, tmp_path,
-    # Insert a batch + remittance + SVC so the Jan-15 visit has a hit.
+):
-    loaded_db.execute(text("""
+    """OA-18-only SVC matches are excluded as DENIED_DUPLICATE_NOISE.
-        INSERT INTO batches (
+
-            id, kind, input_filename, parsed_at
+    Drives the SHIPPED code on the real path: a minimal 835 fixture
-        ) VALUES (
+    in ``tmp_path`` with an OA-18 CAS for member R111111's SVC. R222222
-            'B-TEST-OA18', '835', 'tp-test-20260115.835',
+    has no SVC match → NOT_IN_835. R111111 has a matched SVC whose
-            '2026-01-20 00:00:00'
+    CAS reasons are exactly ``{"OA-18"}`` → DENIED_DUPLICATE_NOISE
-        )
+    → excluded.
-    """))
+    """
-    loaded_db.execute(text("""
+    # Empty CAS → R111111 becomes DENIED (kept), R222222 stays NOT_IN_835.
-        INSERT INTO remittances (
+    no_cas_835 = _build_835_with_svc(
-            id, batch_id, payer_claim_control_number,
+        member_id="R111111",
-            total_charge, total_paid, adjustment_amount,
+        claim_id="CLM-TEST-NOCAS",
-            status_code, received_at, is_reversal
+        status="4",  # denied
-        ) VALUES (
+        procedure="S5150",
-            'R-TEST-OA18', 'B-TEST-OA18', 'CLM-TEST-1',
+        svc_date=date(2026, 1, 15),
-            300.00, 0.00, 0.00,
+        charge=Decimal("300.00"),
-            'DENIED', '2026-01-20 00:00:00', 0
+        paid=Decimal("0.00"),
-        )
+        cas=None,
-    """))
+        pay_date=date(2026, 1, 20),
-    loaded_db.execute(text("""
+    )
-        INSERT INTO service_line_payments (
+    _write_835(tmp_path, "no_cas.835", no_cas_835)
-            remittance_id, line_number, procedure_qualifier,
-            procedure_code, service_date, charge, payment
-        ) VALUES (
-            'R-TEST-OA18', 1, 'HC',
-            'S5150', '2026-01-15', 300.00, 0.00
-        )
-    """))
-    loaded_db.commit()
 
     out = select_spot_check_visits(
         loaded_db,
         window_start=date(2026, 1, 1),
         window_end=date(2026, 6, 27),
         n=10,
+        ingest_dir=tmp_path,
     )
-    # R222222 (no SVC) → NOT_IN_835
-    # R111111 (SVC hit with no CAS yet → DENIED)
     dispositions = sorted(d for (_v, d, _c) in out)
+    # R222222 (no SVC) → NOT_IN_835; R111111 (SVC hit, no CAS yet) → DENIED
     assert dispositions == ["DENIED", "NOT_IN_835"]
 
     # Now add OA-18 CAS — R111111 flips to DENIED_DUPLICATE_NOISE.
-    loaded_db.execute(text("""
+    # Use a different filename so we don't collide with no_cas.835;
-        INSERT INTO cas_adjustments (
+    # the walker re-parses each file independently.
-            remittance_id, group_code, reason_code, amount
+    oa18_835 = _build_835_with_svc(
-        ) VALUES ('R-TEST-OA18', 'OA', '18', 300.00)
+        member_id="R111111",
-    """))
+        claim_id="CLM-TEST-OA18",
-    loaded_db.commit()
+        status="4",  # denied
+        procedure="S5150",
+        svc_date=date(2026, 1, 15),
+        charge=Decimal("300.00"),
+        paid=Decimal("0.00"),
+        cas=["OA-18"],
+        pay_date=date(2026, 1, 20),
+    )
+    _write_835(tmp_path, "oa18.835", oa18_835)
 
     out = select_spot_check_visits(
         loaded_db,
         window_start=date(2026, 1, 1),
         window_end=date(2026, 6, 27),
         n=10,
+        ingest_dir=tmp_path,
     )
     # R111111 now excluded (DENIED_DUPLICATE_NOISE); only R222222 remains.
     assert len(out) == 1
     assert out[0][0].member_id == "R222222"
 
 
+def test_select_spot_check_visits_member_isolation(tmp_path, monkeypatch):
+    """Two visits at the same (procedure, DOS) for DIFFERENT members:
+    member A's SVC has OA-18-only CAS; member B's SVC has none.
+
+    On the correct (member_id, procedure, DOS) indexer, A is excluded
+    and B stays as a candidate. On the previous (procedure, DOS) only
+    indexer, A's CAS would bleed across to B and B would also be
+    excluded. This is the structural fix the verifier demanded.
+    """
+    # Visits for two members at the SAME (procedure, DOS).
+    visits_csv = tmp_path / "visits.csv"
+    with visits_csv.open("w", newline="", encoding="utf-8") as f:
+        writer = csv.writer(f)
+        writer.writerow([
+            "Client", "Visit Date", "Payer", "Authorized", "Member ID",
+            "Auth Start Date", "Auth End Date", "Authorization #", "ICD-10",
+            "Service", "Procedure Code", "Modifiers", "Client Classes",
+            "Billable Hours", "Billable Amount", "Invoice #", "Claimed",
+        ])
+        # Member A at 2026-03-10, $400 — OA-18 candidate (will be excluded).
+        writer.writerow([
+            "Alpha, Anon", "03/10/2026", "COHCPF", "Yes", "R-ALPHA",
+            "01/01/2026", "12/31/2026", "1", "R69",
+            "HCPCS S5150", "S5150", "U8", "DD Waiver", "1", "$400.00",
+            "INV-A", "Yes",
+        ])
+        # Member B at SAME (procedure, DOS), $500 — DENIED candidate (kept).
+        writer.writerow([
+            "Beta, Anon", "03/10/2026", "COHCPF", "Yes", "R-BETA",
+            "01/01/2026", "12/31/2026", "2", "R69",
+            "HCPCS S5150", "S5150", "U8", "DD Waiver", "1", "$500.00",
+            "INV-B", "Yes",
+        ])
+    monkeypatch.setenv("CYCLONE_DB_URL", f"sqlite:///{tmp_path / 'test.db'}")
+    db_mod._reset_for_tests()
+    db_mod.init_db()
+    assert load_visits_csv(visits_csv) == 2
+    SessionLocal = db_mod.SessionLocal()
+    sess = SessionLocal()
+
+    # Two 835s: one for R-ALPHA with OA-18-only CAS, one for R-BETA with no CAS.
+    alpha_835 = _build_835_with_svc(
+        member_id="R-ALPHA",
+        claim_id="CLM-A",
+        status="4",
+        procedure="S5150",
+        svc_date=date(2026, 3, 10),
+        charge=Decimal("400.00"),
+        paid=Decimal("0.00"),
+        cas=["OA-18"],
+        pay_date=date(2026, 3, 15),
+    )
+    beta_835 = _build_835_with_svc(
+        member_id="R-BETA",
+        claim_id="CLM-B",
+        status="4",
+        procedure="S5150",
+        svc_date=date(2026, 3, 10),
+        charge=Decimal("500.00"),
+        paid=Decimal("0.00"),
+        cas=None,
+        pay_date=date(2026, 3, 15),
+    )
+    _write_835(tmp_path, "alpha.835", alpha_835)
+    _write_835(tmp_path, "beta.835", beta_835)
+
+    out = select_spot_check_visits(
+        sess,
+        window_start=date(2026, 1, 1),
+        window_end=date(2026, 6, 27),
+        n=10,
+        ingest_dir=tmp_path,
+    )
+
+    sess.close()
+
+    # Alpha is excluded as DENIED_DUPLICATE_NOISE; Beta is the sole
+    # candidate. n=10 truncates to 1, by billed desc → Beta ($500).
+    assert len(out) == 1
+    assert out[0][0].member_id == "R-BETA"
+    assert out[0][1] == "DENIED"
+
+
 def test_write_spot_check_files_emits_required_segments(tmp_path, loaded_db):
     """Every emitted file passes the plan's literal grep."""
     out_dir = tmp_path / "spot-check-837Ps"
@@ -238,6 +409,7 @@ def test_write_spot_check_files_emits_required_segments(tmp_path, loaded_db):
         window_start=date(2026, 1, 1),
         window_end=date(2026, 6, 27),
         n=10,
+        ingest_dir=tmp_path,
     )
     written = write_spot_check_files(visits, out_dir)
     assert len(written) == 2
@@ -268,6 +440,7 @@ def test_write_spot_check_files_segment_per_line(tmp_path, loaded_db):
         window_start=date(2026, 1, 1),
         window_end=date(2026, 6, 27),
         n=10,
+        ingest_dir=tmp_path,
     )
     written = write_spot_check_files(visits, out_dir)
     # Known segment names that may appear in a well-formed 837P.