diff --git a/backend/src/cyclone/rebill/parse_835_svc.py b/backend/src/cyclone/rebill/parse_835_svc.py
index 1422006..ba3e898 100644
--- a/backend/src/cyclone/rebill/parse_835_svc.py
+++ b/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)
\ No newline at end of file
+    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
\ No newline at end of file
diff --git a/backend/src/cyclone/rebill/run.py b/backend/src/cyclone/rebill/run.py
index 2c5d74d..6bb436c 100644
--- a/backend/src/cyclone/rebill/run.py
+++ b/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
-    #    (._foo.835 are macOS resource forks that the SFTP client surfaces
-    #    alongside real files; matches the convention used by cli, submit-batch,
-    #    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)
+    # 1) Ingest 835s — canonical path through parse_835_svc.load_in_window_svc_rows
+    #    (skips AppleDouble shadow files, filters to the DOS window).
+    svcs = load_in_window_svc_rows(ingest_dir_p, window_start, window_end)
 
     # 2) Load visits CSV — header:
     #    ``Visit Date,Member ID,Procedure Code,Billable Amount``
diff --git a/backend/src/cyclone/rebill/spot_check_pipeline.py b/backend/src/cyclone/rebill/spot_check_pipeline.py
index 80ad0fe..40e4f78 100644
--- a/backend/src/cyclone/rebill/spot_check_pipeline.py
+++ b/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``
-+ ``cas_adjustments`` tables and writes well-formed 837P files to
-disk. The companion validation step lives in
+that consumes the canonical ``visits`` table and the in-window 835
+SVC rows (reparsed from ``ingest/*.x12``) and writes well-formed
+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
-  visits, reconcile against the ingested 835 SVC rows, drop
-  PAID / OA-18-only adjudications, return the top-N candidates by
-  ``billed_amount`` desc.
+- :func:`select_spot_check_visits` — read the in-window visits from
+  the ``visits`` table, reconcile against the 835 SVC rows reparsed
+  from ``ingest_dir`` on the authoritative
+  ``(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``
-  single-claim path. Files are written segment-per-line to ``out_dir``
-  with the HCPF-spec filename via
+  visit via the canonical
+  :func:`cyclone.parsers.serialize_837.serialize_837` single-claim
+  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`` /
-    ``cas_adjustments`` for the same window. Reconciles each visit on
-    ``(member_id, procedure_code, DOS)`` against the SVC rows.
+    Reads in-window visits from ``visits`` and reparses the in-window
+    835 SVC rows from ``ingest_dir`` via the canonical
+    :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_DUPLICATE_NOISE`` — pure OA-18 (Exact Duplicate) CAS.
-        Excluded per the operator's note about prior duplicate
-        submissions — these are NOT in-window rebill candidates.
+      * ``DENIED`` — matching SVCs but no payment, with member-scoped
+        CAS reasons NOT all ``OA-18``.
+      * ``DENIED_DUPLICATE_NOISE`` — pure ``OA-18`` (Exact Duplicate)
+        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(
-        db_mod.text(
-            """
-            SELECT slp.procedure_code AS procedure,
-                   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()
+    # Build parallel lists so the outcome index aligns with the
+    # visit index — ``reconcile_visits_to_835`` preserves order.
+    spot_rows: list[VisitRow] = [_spot_check_visit_row_from_db(v) for v in db_visits]
+    recon_rows: list[ReconcileVisitRow] = [_reconcile_visit_row_from_db(v) for v in db_visits]
 
-    cas_by_remit: dict[str, list[str]] = {}
-    cas_rows = session.execute(
-        db_mod.text(
-            """
-            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}"
-        )
+    ingest_dir_p = Path(ingest_dir) if ingest_dir is not None else Path("ingest")
+    svc_rows = load_in_window_svc_rows(ingest_dir_p, window_start, window_end)
+    cas_lookup = _member_scoped_cas_lookup(svc_rows)
 
-    # SQLite returns DATE columns as ISO strings; normalize to date.
-    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)
+    outcomes: list[ReconcileOutcome] = reconcile_visits_to_835(recon_rows, svc_rows)
 
     out: list[tuple[VisitRow, str, list[str]]] = []
-    for v in visits:
-        vr = _visit_row_from_db(v)
-        cands = by_pdos.get((vr.procedure_code, vr.dos), [])
-        if not cands:
+    for vr, outcome in zip(spot_rows, outcomes):
+        if outcome.category == OutcomeCategory.NOT_IN_835:
             out.append((vr, "NOT_IN_835", []))
             continue
-        total_paid = sum(
-            (Decimal(str(c.paid or 0)) for c in cands), Decimal("0")
+        # DENIED / PAID / PARTIAL all had a matched SVC. CAS reasons
+        # 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)
-        cas_reasons: list[str] = []
-        for c in cands:
-            cas_reasons.extend(cas_by_remit.get(str(c.remittance_id), []))
-        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))
+        if outcome.category == OutcomeCategory.PAID:
+            out.append((vr, "PAID", matched_cas))
+        elif outcome.category == OutcomeCategory.PARTIAL:
+            out.append((vr, "PARTIAL", matched_cas))
         else:
-            if cas_reasons and set(cas_reasons) == {"OA-18"}:
-                out.append((vr, "DENIED_DUPLICATE_NOISE", cas_reasons))
-            else:
-                out.append((vr, "DENIED", cas_reasons))
+            out.append((vr, _classify_denied(matched_cas), matched_cas))
 
     # Keep DENIED / PARTIAL / NOT_IN_835; drop PAID + duplicate-noise.
     candidates = [(v, d, c) for (v, d, c) in out
diff --git a/backend/tests/test_spot_check_pipeline.py b/backend/tests/test_spot_check_pipeline.py
index f0d6509..0bef76a 100644
--- a/backend/tests/test_spot_check_pipeline.py
+++ b/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):
-    """With no SVC rows in DB, visits are NOT_IN_835 (rebill candidates)."""
+def test_select_spot_check_visits_marks_not_in_835(loaded_db, tmp_path):
+    """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):
-    """OA-18-only SVC matches are excluded as DENIED_DUPLICATE_NOISE."""
-    # Insert a batch + remittance + SVC so the Jan-15 visit has a hit.
-    loaded_db.execute(text("""
-        INSERT INTO batches (
-            id, kind, input_filename, parsed_at
-        ) VALUES (
-            'B-TEST-OA18', '835', 'tp-test-20260115.835',
-            '2026-01-20 00:00:00'
-        )
-    """))
-    loaded_db.execute(text("""
-        INSERT INTO remittances (
-            id, batch_id, payer_claim_control_number,
-            total_charge, total_paid, adjustment_amount,
-            status_code, received_at, is_reversal
-        ) VALUES (
-            'R-TEST-OA18', 'B-TEST-OA18', 'CLM-TEST-1',
-            300.00, 0.00, 0.00,
-            'DENIED', '2026-01-20 00:00:00', 0
-        )
-    """))
-    loaded_db.execute(text("""
-        INSERT INTO service_line_payments (
-            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()
+def test_select_spot_check_visits_oa18_excluded_member_scoped(
+    loaded_db, tmp_path,
+):
+    """OA-18-only SVC matches are excluded as DENIED_DUPLICATE_NOISE.
+
+    Drives the SHIPPED code on the real path: a minimal 835 fixture
+    in ``tmp_path`` with an OA-18 CAS for member R111111's SVC. R222222
+    has no SVC match → NOT_IN_835. R111111 has a matched SVC whose
+    CAS reasons are exactly ``{"OA-18"}`` → DENIED_DUPLICATE_NOISE
+    → excluded.
+    """
+    # Empty CAS → R111111 becomes DENIED (kept), R222222 stays NOT_IN_835.
+    no_cas_835 = _build_835_with_svc(
+        member_id="R111111",
+        claim_id="CLM-TEST-NOCAS",
+        status="4",  # denied
+        procedure="S5150",
+        svc_date=date(2026, 1, 15),
+        charge=Decimal("300.00"),
+        paid=Decimal("0.00"),
+        cas=None,
+        pay_date=date(2026, 1, 20),
+    )
+    _write_835(tmp_path, "no_cas.835", no_cas_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,
     )
-    # 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("""
-        INSERT INTO cas_adjustments (
-            remittance_id, group_code, reason_code, amount
-        ) VALUES ('R-TEST-OA18', 'OA', '18', 300.00)
-    """))
-    loaded_db.commit()
+    # Use a different filename so we don't collide with no_cas.835;
+    # the walker re-parses each file independently.
+    oa18_835 = _build_835_with_svc(
+        member_id="R111111",
+        claim_id="CLM-TEST-OA18",
+        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.