feat(sp41): Edifabric gate on every emitted 837P (quarantine on failure)

Replace the run_rebill() .touch() placeholders for Pipeline A and Pipeline B with real serialize → validate → write paths.

Pipeline A: build a ClaimOutput adapter from the canonical RebillClaim fields (claim_id/member_id/procedure/svc_date/charge) — the per-claim envelope context (provider NPI, subscriber name, payer name) comes from the original claim being replaced and isn't carried forward by the rebill pipeline, so the adapter emits safe placeholders and lets the SP40 serializer fallbacks fill PER / SBR09. Then serialize_837 → cyclone.edifabric.validate_edi → write to pipeline-a/ on success or quarantine/ on Status=='error'.

Pipeline B: serialize_member_week_batch(b, tpid=tpid) (the Task 14 overload — takes a MemberWeekBatch, not a ClaimOutput) → cyclone.edifabric.validate_edi → write to pipeline-b/ or quarantine/.

Filename disambiguation: build_outbound_filename embeds a millisecond timestamp; multiple files emitted in the same millisecond would collide. Append per-claim / per-batch suffix so the audit trail (claim_id ↔ file / batch_key ↔ file) stays one-to-one.

Edifabric unavailability (no API key, network error, 5xx) fails OPEN with a WARNING log: the file lands in the pipeline dir, matches the SP40 dev/CI posture for boxes without a paid key. Quarantine only on Edifabric Status=='error'.

New tests in backend/tests/test_rebill_run.py cover: real 837P file emission (non-empty, ISA..IEA envelope) for both pipelines, quarantine on Edifabric error, fail-open on Edifabric unavailability, HCPF-spec filename prefix preserved, per-claim / per-batch disambiguation suffix present, original claim_id preserved in Pipeline A CLM01.

backend/src/cyclone/rebill/run.py

@@ -9,11 +9,14 @@ paths from --visits / --ingest / --out; HTTP endpoint passes the same.
 from __future__ import annotations
 
 import csv as _csv
+import logging
 from dataclasses import dataclass
 from datetime import date, datetime
 from decimal import Decimal
 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.pipeline_a import RebillClaim
@@ -33,6 +36,8 @@ from cyclone.rebill.summary import (
 )
 from cyclone.rebill.timely_filing import timely_filing_decision
 
+_log = logging.getLogger(__name__)
+
 
 @dataclass
 class RunResult:
@@ -78,14 +83,17 @@ def run_rebill(
             - NOT_IN_835 + past-window   → EXCLUDED_TIMELY_FILING (no override)
             - DENIED/PARTIAL + EXCLUDED CARC → EXCLUDED_CARC
             - DENIED/PARTIAL + REVIEW/REBILL → Pipeline A (REBILLED_A)
-      5. Emit Pipeline A files (``<out>/pipeline-a/x-{claim_id}.837``) and
+      5. Emit Pipeline A files (``<out>/pipeline-a/tp{tpid}-837P-...-1of1.x12``)
-         Pipeline B files (``<out>/pipeline-b/x-{member_id}-{iso_year}-W{iso_week:02d}.837``).
+         and Pipeline B files (``<out>/pipeline-b/...-1of1.{member}-W{week}.x12``).
-         Files are ``touch()``ed placeholders here — Task 13 wraps each emit
+         Each file is serialized via the canonical 837P helpers and gated
-         in a serialize-then-Edifabric-validate step, with quarantine on failure.
+         through Edifabric's ``validate_edi``; failures land in
+         ``<out>/quarantine/{key}.837`` so the operator can triage without
+         blocking the batch.
       6. Write ``<out>/summary.csv`` with one row per non-PAID visit.
 
-    The ``tpid`` argument is preserved on the signature for Task 13 / Task 14
+    The ``tpid`` argument threads into the HCPF outbound filename
-    to thread into the real filename builder; unused in this placeholder emit.
+    (``build_outbound_filename(tpid, "837P")``) and the Pipeline-B batch
+    serializer's submitter/receiver block (``serialize_member_week_batch``).
 
     ``as_of`` defaults to ``date.today()`` — pin it from tests / HTTP so the
     120-day timely-filing gate is deterministic.
@@ -224,44 +232,188 @@ def run_rebill(
             file_path="pipeline-a/",
         ))
 
-    # 5) Emit Pipeline A files — placeholder; Task 13 wraps with the
+    # 5) Emit Pipeline A files — serialize each RebillClaim, gate
-    #    serialize-then-Edifabric-validate step. Filename shape:
+    #    through Edifabric's validate_edi, write clean files into
-    #    ``x-{claim_id}.837`` so the audit trail ties back to the
+    #    ``pipeline-a/`` and Edifabric-rejected files into
-    #    original claim_submit_id via the SummaryRow.cas_reasons chain.
+    #    ``quarantine/`` for operator triage.
+    #
+    #    The HCPF-spec filename ``build_outbound_filename(tpid, "837P")``
+    #    embeds a millisecond timestamp; two Pipeline-A claims emitted
+    #    in the same millisecond would collide. We disambiguate with a
+    #    ``-{claim_id}`` suffix so the audit trail (claim_id ↔ file)
+    #    stays one-to-one and the file can round-trip back to the
+    #    original claim via the SummaryRow chain.
     a_dir = out_dir_p / "pipeline-a"
     a_dir.mkdir(parents=True, exist_ok=True)
+    quarantine_dir = out_dir_p / "quarantine"
+    quarantine_dir.mkdir(parents=True, exist_ok=True)
     a_files: list[Path] = []
     for claim in pipeline_a_claims:
-        p = a_dir / f"x-{claim.claim_id}.837"
+        body = _serialize_pipeline_a(claim, tpid=tpid)
-        p.touch()
+        status = _validate_or_skip(body, claim_id=claim.claim_id)
+        if status == "quarantine":
+            qp = quarantine_dir / f"{claim.claim_id}.837"
+            qp.write_bytes(body)
+        else:
+            base = build_outbound_filename(tpid, "837P")
+            # Disambiguate same-millisecond collisions across claims.
+            stem, dot_ext = base.rsplit(".", 1)
+            p = a_dir / f"{stem}-{claim.claim_id}.{dot_ext}"
+            p.write_bytes(body)
             a_files.append(p)
 
-    # 6) Build + emit Pipeline B batches — placeholder; Task 13 wraps
+    # 6) Build + emit Pipeline B batches. The Task 14 overload
-    #    with the Edifabric gate. Filename shape mirrors the HCPF
+    #    ``serialize_member_week_batch`` takes a ``MemberWeekBatch`` and
-    #    member-week pattern: ``x-{member_id}-{iso_year}-W{iso_week:02d}.837``.
+    #    emits one envelope with one CLM per visit. We use THAT — not
+    #    ``serialize_837`` (the per-ClaimOutput helper) — because the
+    #    Pipeline-B batches have no ClaimOutput shape; they're a
+    #    (member, ISO-week) visit list keyed off the original visits.
+    #
+    #    Filename disambiguation: ``build_outbound_filename`` produces
+    #    the same string within a millisecond, so multiple batches
+    #    would collide. Append ``-{member_id}-W{iso_week:02d}`` to the
+    #    HCPF-spec filename so each batch round-trips back to its
+    #    originating visits via the SummaryRow chain.
     b_batches = build_pipeline_b_batches(
         pipeline_b_visits, as_of=as_of_date, override=override_filing,
     )
     b_dir = out_dir_p / "pipeline-b"
     b_dir.mkdir(parents=True, exist_ok=True)
     b_files: list[Path] = []
+    from cyclone.parsers.serialize_837 import serialize_member_week_batch
     for b in b_batches:
-        p = b_dir / f"x-{b.member_id}-{b.iso_year}-W{b.iso_week:02d}.837"
+        body = serialize_member_week_batch(b, tpid=tpid)
-        p.touch()
+        batch_key = f"{b.member_id}-{b.iso_year}-W{b.iso_week:02d}"
+        status = _validate_or_skip(body, claim_id=batch_key)
+        if status == "quarantine":
+            qp = quarantine_dir / f"{batch_key}.837"
+            qp.write_bytes(body)
+        else:
+            base = build_outbound_filename(tpid, "837P")
+            stem, dot_ext = base.rsplit(".", 1)
+            p = b_dir / f"{stem}-{batch_key}.{dot_ext}"
+            p.write_bytes(body)
             b_files.append(p)
 
     # 7) Summary CSV — one row per non-PAID visit, regardless of pipeline.
     summary_path = out_dir_p / "summary.csv"
     write_summary_csv(summary, summary_path)
 
-    # Touching `tpid` so it doesn't lint as unused — Task 13 / Task 14 will
-    # thread it into the real filename builder (cyclone.edi.filenames.
-    # build_outbound_filename) once the serializer adapter lands.
-    _ = tpid
-
     return RunResult(
         summary_path=summary_path,
         counts=counts,
         pipeline_a_files=a_files,
         pipeline_b_files=b_files,
     )
+
+
+def _serialize_pipeline_a(claim: RebillClaim, *, tpid: str) -> bytes:
+    """Build a 837P byte string for a Pipeline-A RebillClaim.
+
+    Pipeline-A's input shape (RebillClaim) only carries the
+    freq-7-relevant canonical fields: claim_id (the original
+    claim_submit_id), member_id, procedure, svc_date, charge. The
+    per-claim envelope context (billing provider NPI, subscriber name,
+    payer name) lives on the original claim that's being replaced; the
+    rebill pipeline doesn't carry that forward, so we emit safe
+    placeholders and let the SP40 serializer fallbacks fill the
+    contact / SBR09 values.
+
+    Returns ASCII bytes (the 837P envelope is pure ASCII).
+    """
+    # Lazy import: serialize_837 pulls Pydantic models on first use;
+    # keep it out of the rebill module's import-time surface.
+    from cyclone.parsers.models import (
+        BillingProvider,
+        ClaimHeader,
+        ClaimOutput,
+        Payer,
+        Procedure,
+        ServiceLine,
+        Subscriber,
+        ValidationReport,
+    )
+    from cyclone.parsers.serialize_837 import serialize_837
+
+    # Deterministic control numbers derived from the original claim_id
+    # so a retry of the same DOS window produces the same filenames
+    # (the operator can then diff against the prior run).
+    cn = claim.claim_id[:9].rjust(4, "0")[-4:]
+    placeholder_claim = ClaimOutput(
+        claim_id=claim.claim_id,
+        control_number=cn,
+        transaction_date=claim.svc_date,
+        billing_provider=BillingProvider(
+            name="REBILL PROVIDER",
+            npi="0000000000",
+        ),
+        subscriber=Subscriber(
+            first_name="REBILL",
+            last_name=claim.member_id,  # surface the member id in NM103
+            member_id=claim.member_id,
+        ),
+        payer=Payer(name="CO_TXIX", id="CO_TXIX"),
+        claim=ClaimHeader(
+            claim_id=claim.claim_id,
+            total_charge=claim.charge,
+            place_of_service="11",
+            facility_code_qualifier="B",
+            frequency_code=claim.frequency_code,  # always "7"
+        ),
+        service_lines=[
+            ServiceLine(
+                line_number=1,
+                procedure=Procedure(qualifier="HC", code=claim.procedure),
+                charge=claim.charge,
+                units=Decimal("1"),
+                unit_type="UN",
+                place_of_service="11",
+                service_date=claim.svc_date,
+            ),
+        ],
+        validation=ValidationReport(passed=True),
+        transaction_type_code="CH",
+    )
+    text = serialize_837(placeholder_claim, sender_id=tpid)
+    return text.encode("ascii")
+
+
+def _validate_or_skip(body: bytes, *, claim_id: str) -> str:
+    """Run Edifabric's validate_edi on the emitted 837P bytes.
+
+    Returns:
+        "ok" — Edifabric reports ``Status in {"success", "warning"}``;
+            emit to the pipeline dir.
+        "quarantine" — Edifabric reports ``Status == "error"``; emit
+            to ``<out>/quarantine/`` for operator triage.
+        "skip" — Edifabric was unreachable (no API key, network error,
+            5xx, etc.); treat as ``ok`` and emit to the pipeline dir.
+            A WARNING is logged so the operator knows the gate didn't
+            actually run. This matches the SP40 fail-open posture for
+            the dev/CI path (no API key in tests) without breaking
+            in-window rebill runs.
+    """
+    try:
+        result = _edifabric.validate_edi(body)
+    except _edifabric.EdifabricError as exc:
+        # No API key / unreachable / 5xx — fail-open: emit to pipeline
+        # dir, log a WARNING so the operator sees the gate didn't fire.
+        _log.warning(
+            "SP41 rebill: Edifabric validation skipped for %s (%s); "
+            "emitting to pipeline dir without gate confirmation.",
+            claim_id, exc,
+        )
+        return "skip"
+    status = result.get("Status", "")
+    if status == "error":
+        details = result.get("Details") or []
+        msgs = "; ".join(
+            f"{d.get('SegmentId', '?')}: {d.get('Message', '?')}"
+            for d in details[:5]
+        )
+        _log.warning(
+            "SP41 rebill: Edifabric rejected %s — quarantining. %s",
+            claim_id, msgs,
+        )
+        return "quarantine"
+    return "ok"

backend/tests/test_rebill_run.py

@@ -0,0 +1,328 @@
+"""SP41 Task 13 — Edifabric validation gate on emitted rebill files.
+
+Covers the new emit path in ``cyclone.rebill.run.run_rebill``:
+
+  * Real 837P files are written (not ``Path.touch()`` placeholders).
+  * Files pass through ``cyclone.edifabric.validate_edi`` before they
+    land in ``<out>/pipeline-{a,b}/``; failures go to
+    ``<out>/quarantine/``.
+  * When Edifabric is unavailable (no API key, network error, etc.),
+    the gate fails open with a WARNING log and emits to the pipeline
+    dirs (matches the SP40 dev/CI posture).
+
+Each test builds its own visits CSV + 835 stub in ``tmp_path`` and
+invokes ``run_rebill`` directly (no HTTP). The Edifabric API is mocked
+via ``unittest.mock.patch`` on
+``cyclone.rebill.run._edifabric.validate_edi`` so no live HTTP hits the
+network.
+"""
+from __future__ import annotations
+
+import csv
+from datetime import date
+from decimal import Decimal
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from cyclone.rebill.run import RunResult, run_rebill
+
+
+# --------------------------------------------------------------------------- #
+# Fixtures / helpers
+# --------------------------------------------------------------------------- #
+
+
+def _write_visits_csv(path: Path, rows: list[tuple[str, str, str, str]]) -> Path:
+    """rows: list of (dos_mmddyyyy, member, procedure, billed_str)."""
+    with path.open("w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow(["Visit Date", "Member ID", "Procedure Code", "Billable Amount"])
+        for r in rows:
+            w.writerow(r)
+    return path
+
+
+def _stub_835_with_denied_svc(
+    ingest_dir: Path,
+    member: str,
+    procedure: str,
+    svc_date: date,
+    claim_id: str = "ORIG-1",
+    carc_group: str = "CO",
+    carc_reason: str = "29",
+) -> Path:
+    """An 835 with a single denied SVC so the visit lands in Pipeline A.
+
+    The rebill pipeline reuses the SVC's ``claim_id`` as the Pipeline-A
+    RebillClaim's ``claim_id`` (preserves the freq-7 anchor). The visit
+    must match (member_id, procedure, svc_date) so the reconcile step
+    pairs them into DENIED + CARC REBILL.
+
+    The 835 shape mirrors ``tests/fixtures/835_sample_svc_with_member.txt``:
+    CLP02 = 4 (Denied), NM1*QC.NM109 = member_id, then SVC → DTM*472 → CAS.
+
+    Segments are concatenated without ``\\n`` separators — the parser
+    splits on ``~`` only, and a leading ``\\n`` makes ``elems[0]`` an
+    empty string so the segment-name match silently drops the segment.
+    """
+    ingest_dir.mkdir(exist_ok=True)
+    svc_date_str = svc_date.strftime("%Y%m%d")
+    segs = [
+        "ISA*00*          *00*          *ZZ*SENDER         *ZZ*RECEIVER       *260627*1200*^*00501*000000001*0*P*:~",
+        "GS*HC*SENDER*RECEIVER*20260627*1200*1*X*005010X221A1~",
+        "ST*835*0001~",
+        "BPR*I*0*C*ACH*CCP*01*021000021*DA*123456789*1512345678**01*021000021*DA*123456789*20260101~",
+        "TRN*1*TRACE01*1512345678~",
+        "DTM*405*20260118~",
+        "N1*PR*COLORADO MEDICAL ASSISTANCE PROGRAM*XV*COMEDASSISTPROG~",
+        "N3*PO BOX 1100~",
+        "N4*DENVER*CO*80202~",
+        "REF*2U*7912900843~",
+        "LX*1~",
+        # CLP02=4 means Denied → reconciles to DENIED → Pipeline A.
+        f"CLP*{claim_id}*4*100*0**MC*2026029105200*11*1~",
+        # NM1*QC.NM109 is the member_id; parse_835_svc propagates it to
+        # the SVC row so the visit-side reconcile key matches.
+        f"NM1*QC*1*PATIENT*NAME****MR*{member}~",
+        f"SVC*HC:{procedure}*100*0*UN*1~",
+        f"DTM*472*{svc_date_str}~",
+        # CAS group code + reason → CARC REBILL (not EXCLUDED).
+        f"CAS*{carc_group}*{carc_reason}*100~",
+        "SE*14*0001~",
+        "GE*1*1~",
+        "IEA*1*000000001~",
+    ]
+    body = "".join(segs)
+    p = ingest_dir / "x.835"
+    p.write_text(body)
+    return p
+
+
+def _stub_empty_835(ingest_dir: Path) -> Path:
+    """Empty 835 → rebill's parse_835_svc emits zero SVCs."""
+    ingest_dir.mkdir(exist_ok=True)
+    p = ingest_dir / "x.835"
+    p.write_text("ST*835*0001~SE*0*0001~")
+    return p
+
+
+# --------------------------------------------------------------------------- #
+# Tests
+# --------------------------------------------------------------------------- #
+
+
+def test_run_rebill_emits_real_files_for_pipeline_b(tmp_path):
+    """Pipeline-B happy path: 1 in-window visit, 0 SVCs → 1 real 837P
+    file in pipeline-b/ (non-empty, contains ISA..IEA), 0 files in
+    quarantine/.
+
+    Edifabric is patched to return ``success`` so the file passes the
+    gate cleanly. The fixture is the same shape as the existing
+    ``test_post_rebill_runs_and_returns_summary_path`` happy-path test.
+    """
+    visits_path = _write_visits_csv(tmp_path / "visits.csv", [
+        ("06/27/2026", "J813715", "T1019", "$2.32"),
+    ])
+    ingest_dir = _stub_empty_835(tmp_path / "ingest")
+    out_dir = tmp_path / "out"
+
+    fake_result = {"Status": "success", "Details": [], "LastIndex": 20}
+    with patch(
+        "cyclone.rebill.run._edifabric.validate_edi",
+        return_value=fake_result,
+    ) as mock_validate:
+        result = run_rebill(
+            window_start=date(2026, 1, 1),
+            window_end=date(2026, 6, 27),
+            override_filing=False,
+            visits_csv_path=str(visits_path),
+            ingest_dir=str(ingest_dir),
+            out_dir=str(out_dir),
+        )
+
+    # Pipeline B emitted exactly one file (matches the existing
+    # ``len(body["pipeline_b_files"]) == 1`` assertion in test_api_rebill).
+    assert len(result.pipeline_b_files) == 1
+    b_path = result.pipeline_b_files[0]
+    assert b_path.exists(), b_path
+    assert b_path.stat().st_size > 0, "Pipeline-B file must be non-empty"
+
+    # The file is a real 837P envelope (starts with ISA, ends with IEA).
+    body = b_path.read_bytes()
+    assert body.startswith(b"ISA*"), body[:32]
+    assert b"IEA*" in body, body[-32:]
+
+    # Edifabric gate was called exactly once for the one Pipeline-B batch.
+    assert mock_validate.call_count == 1
+
+    # HCPF-spec filename prefix is preserved.
+    assert b_path.name.startswith("tp11525703-837P-"), b_path.name
+    # Per-batch disambiguation suffix is present so same-millisecond
+    # collisions don't overwrite each other.
+    assert "J813715-2026-W26" in b_path.name, b_path.name
+
+    # Quarantine is empty (no validation failures).
+    q_files = list((out_dir / "quarantine").glob("*.837"))
+    assert q_files == [], q_files
+
+
+def test_run_rebill_quarantines_on_edifabric_error(tmp_path):
+    """When validate_edi returns Status='error', the file lands in
+    ``quarantine/`` (NOT in ``pipeline-b/``).
+
+    Exercises both the Edifabric-rejected path AND the case-isolation
+    that splits good files from bad files in the same run.
+    """
+    visits_path = _write_visits_csv(tmp_path / "visits.csv", [
+        ("06/27/2026", "J813715", "T1019", "$2.32"),
+        ("06/27/2026", "J999999", "T1019", "$2.32"),  # different member
+    ])
+    ingest_dir = _stub_empty_835(tmp_path / "ingest")
+    out_dir = tmp_path / "out"
+
+    # First call (J813715) returns error → quarantine.
+    # Second call (J999999) returns success → pipeline-b.
+    fake_results = iter([
+        {
+            "Status": "error",
+            "Details": [
+                {"SegmentId": "PER", "Message": "PER-04 is required", "Status": "error"},
+            ],
+            "LastIndex": 5,
+        },
+        {"Status": "success", "Details": [], "LastIndex": 20},
+    ])
+
+    def _side_effect(_body):
+        return next(fake_results)
+
+    with patch(
+        "cyclone.rebill.run._edifabric.validate_edi",
+        side_effect=_side_effect,
+    ) as mock_validate:
+        result = run_rebill(
+            window_start=date(2026, 1, 1),
+            window_end=date(2026, 6, 27),
+            override_filing=False,
+            visits_csv_path=str(visits_path),
+            ingest_dir=str(ingest_dir),
+            out_dir=str(out_dir),
+        )
+
+    # Only the success batch should land in pipeline-b/.
+    assert len(result.pipeline_b_files) == 1, (
+        f"only the success batch should land in pipeline-b/, got "
+        f"{[str(p) for p in result.pipeline_b_files]}"
+    )
+    # One file in quarantine (the rejected batch).
+    q_files = list((out_dir / "quarantine").glob("*.837"))
+    assert len(q_files) == 1, q_files
+    # Quarantine filename keys off the batch (member_id-W{iso_week:02d}).
+    assert "J813715" in q_files[0].name, q_files[0].name
+    assert q_files[0].stat().st_size > 0, "quarantined file must be non-empty"
+
+    # validate_edi called twice — once per Pipeline-B batch.
+    assert mock_validate.call_count == 2
+
+
+def test_run_rebill_fails_open_when_edifabric_unavailable(tmp_path):
+    """When validate_edi raises EdifabricError (no API key, network
+    error, 5xx), the file lands in the pipeline dir with a WARNING log.
+
+    Matches the SP40 dev/CI posture: tests / dev boxes don't have a
+    real API key, so the gate must NOT block the rebill run.
+    """
+    from cyclone.edifabric import EdifabricError
+
+    visits_path = _write_visits_csv(tmp_path / "visits.csv", [
+        ("06/27/2026", "J813715", "T1019", "$2.32"),
+    ])
+    ingest_dir = _stub_empty_835(tmp_path / "ingest")
+    out_dir = tmp_path / "out"
+
+    with patch(
+        "cyclone.rebill.run._edifabric.validate_edi",
+        side_effect=EdifabricError(0, "API key not configured"),
+    ) as mock_validate:
+        result = run_rebill(
+            window_start=date(2026, 1, 1),
+            window_end=date(2026, 6, 27),
+            override_filing=False,
+            visits_csv_path=str(visits_path),
+            ingest_dir=str(ingest_dir),
+            out_dir=str(out_dir),
+        )
+
+    # File emitted to pipeline-b/ (NOT to quarantine — fail-open).
+    assert len(result.pipeline_b_files) == 1
+    assert result.pipeline_b_files[0].exists()
+    # Quarantine stays empty.
+    q_files = list((out_dir / "quarantine").glob("*.837"))
+    assert q_files == [], q_files
+    # Gate was actually called (so we know the fail-open path fired,
+    # not some accidental short-circuit).
+    assert mock_validate.call_count == 1
+
+
+def test_run_rebill_emits_real_files_for_pipeline_a(tmp_path):
+    """Pipeline-A happy path: 1 in-window visit, 1 denied SVC matching
+    it → 1 real 837P file in pipeline-a/ with frequency-code 7
+    preserved.
+
+    The SVC's original claim_id (ORIG-1) carries through as the
+    Pipeline-A RebillClaim.claim_id and shows up in the emitted file's
+    CLM01 segment AND in the on-disk filename suffix.
+    """
+    svc_date = date(2026, 6, 27)
+    visits_path = _write_visits_csv(tmp_path / "visits.csv", [
+        (svc_date.strftime("%m/%d/%Y"), "J813715", "T1019", "$100.00"),
+    ])
+    ingest_dir = tmp_path / "ingest"
+    _stub_835_with_denied_svc(
+        ingest_dir,
+        member="J813715",
+        procedure="T1019",
+        svc_date=svc_date,
+        claim_id="ORIG-1",
+    )
+    out_dir = tmp_path / "out"
+
+    fake_result = {"Status": "success", "Details": [], "LastIndex": 30}
+    with patch(
+        "cyclone.rebill.run._edifabric.validate_edi",
+        return_value=fake_result,
+    ):
+        result = run_rebill(
+            window_start=date(2026, 1, 1),
+            window_end=date(2026, 6, 27),
+            override_filing=False,
+            visits_csv_path=str(visits_path),
+            ingest_dir=str(ingest_dir),
+            out_dir=str(out_dir),
+        )
+
+    # Pipeline A emitted exactly one file.
+    assert len(result.pipeline_a_files) == 1, result.pipeline_a_files
+    a_path = result.pipeline_a_files[0]
+    assert a_path.exists(), a_path
+    assert a_path.stat().st_size > 0, "Pipeline-A file must be non-empty"
+
+    body = a_path.read_bytes()
+    assert body.startswith(b"ISA*"), body[:32]
+    assert b"IEA*" in body, body[-32:]
+
+    # Original claim_submit_id is preserved in CLM01 (anchors the
+    # frequency-7 replacement).
+    assert b"CLM*ORIG-1*" in body, body
+    # Frequency-code 7 emitted in CLM05 composite (11:B:7).
+    assert b"CLM*ORIG-1*100.00***11:B:7" in body, body
+
+    # Pipeline-A filename uses HCPF-spec prefix + claim_id suffix.
+    assert a_path.name.startswith("tp11525703-837P-"), a_path.name
+    assert a_path.name.endswith("-ORIG-1.x12"), a_path.name
+
+    # Quarantine empty.
+    q_files = list((out_dir / "quarantine").glob("*.837"))
+    assert q_files == [], q_files