diff --git a/backend/src/cyclone/rebill/run.py b/backend/src/cyclone/rebill/run.py
new file mode 100644
index 0000000..beef89d
--- /dev/null
+++ b/backend/src/cyclone/rebill/run.py
@@ -0,0 +1,267 @@
+"""Top-level orchestrator for SP41.
+
+Wires the 835 SVC reparse → reconciliation → CARC filter → timely-filing
+gate → pipeline A → pipeline B → summary CSV → Edifabric validation.
+
+The CLI and the HTTP endpoint both call this. CLI passes filesystem
+paths from --visits / --ingest / --out; HTTP endpoint passes the same.
+"""
+from __future__ import annotations
+
+import csv as _csv
+from dataclasses import dataclass
+from datetime import date, datetime
+from decimal import Decimal
+from pathlib import Path
+
+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
+from cyclone.rebill.pipeline_b import build_pipeline_b_batches
+from cyclone.rebill.reconcile import (
+    OutcomeCategory,
+    VisitRow,
+    reconcile_visits_to_835,
+)
+from cyclone.rebill.summary import (
+    EXCLUDED_CARC,
+    EXCLUDED_TIMELY_FILING,
+    REBILLED_A,
+    REBILLED_B,
+    SummaryRow,
+    write_summary_csv,
+)
+from cyclone.rebill.timely_filing import timely_filing_decision
+
+
+@dataclass
+class RunResult:
+    """The return value of run_rebill — where the summary CSV landed, the
+    per-disposition counts, and the list of pipeline A / B file paths.
+
+    Mutable (no frozen=True) because the contained `counts` dict and
+    `pipeline_*_files` lists are mutated by callers who aggregate results
+    across runs.
+    """
+
+    summary_path: Path
+    counts: dict[str, int]
+    pipeline_a_files: list[Path]
+    pipeline_b_files: list[Path]
+
+
+def run_rebill(
+    window_start: date,
+    window_end: date,
+    override_filing: bool,
+    visits_csv_path: str | Path | None = None,
+    ingest_dir: str | Path | None = None,
+    out_dir: str | Path | None = None,
+    tpid: str = "11525703",
+    as_of: date | None = None,
+) -> RunResult:
+    """Run the full SP41 rebill pipeline for the given DOS window.
+
+    Pipeline shape:
+
+      1. Walk ``ingest_dir`` for *.835 / *.x12 (skipping AppleDouble shadow
+         files matching ``._*``), reparse each into SvcRows via
+         ``parse_835_svc``, keep those with svc_date in [window_start, window_end].
+      2. Load the AxisCare visits CSV (``visits_csv_path``) and keep the rows
+         whose Visit Date is in the same DOS window.
+      3. Reconcile visits against the in-window SVCs on the
+         (member_id, procedure, DOS) match key. Best-of-N: PAID if
+         total_paid >= 95% of billed, else PARTIAL if any payment, else DENIED.
+      4. Per-visit disposition:
+            - PAID                       → skip (not in summary)
+            - NOT_IN_835 + within-window → Pipeline B (REBILLED_B)
+            - 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
+         Pipeline B files (``<out>/pipeline-b/x-{member_id}-{iso_year}-W{iso_week:02d}.837``).
+         Files are ``touch()``ed placeholders here — Task 13 wraps each emit
+         in a serialize-then-Edifabric-validate step, with quarantine on failure.
+      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
+    to thread into the real filename builder; unused in this placeholder emit.
+
+    ``as_of`` defaults to ``date.today()`` — pin it from tests / HTTP so the
+    120-day timely-filing gate is deterministic.
+    """
+    as_of_date = as_of or date.today()
+    out_dir_p = Path(out_dir or f"dev/rebills/{date.today().isoformat()}")
+    out_dir_p.mkdir(parents=True, exist_ok=True)
+    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):
+            if window_start <= s.svc_date <= window_end:
+                svcs.append(s)
+
+    # 2) Load visits CSV — header:
+    #    ``Visit Date,Member ID,Procedure Code,Billable Amount``
+    #    DOS is MM/DD/YYYY; Billable Amount may carry a ``$`` prefix and
+    #    ``,`` thousands separators.
+    visits: list[VisitRow] = []
+    with visits_csv_p.open(newline="") as f:
+        for r in _csv.DictReader(f):
+            dos = datetime.strptime(r["Visit Date"].strip(), "%m/%d/%Y").date()
+            if not (window_start <= dos <= window_end):
+                continue
+            amt = Decimal(
+                r["Billable Amount"].strip().lstrip("$").replace(",", "")
+            )
+            visits.append(VisitRow(
+                date=dos,
+                member_id=r["Member ID"].strip(),
+                procedure=r["Procedure Code"].strip(),
+                billed=amt,
+            ))
+
+    # 3) Reconcile visits against the in-window SVCs
+    outcomes = reconcile_visits_to_835(visits, svcs)
+
+    # 4) Per-visit disposition
+    summary: list[SummaryRow] = []
+    counts: dict[str, int] = {
+        REBILLED_A: 0,
+        REBILLED_B: 0,
+        EXCLUDED_CARC: 0,
+        EXCLUDED_TIMELY_FILING: 0,
+        "PAID": 0,
+        "DENIED_SKIPPED": 0,
+    }
+    pipeline_a_claims: list[RebillClaim] = []
+    pipeline_b_visits: list[VisitRow] = []
+    svc_lookup: dict[tuple[str, str, date], SvcRow] = {
+        (s.member_id, s.procedure, s.svc_date): s for s in svcs
+    }
+
+    for o in outcomes:
+        if o.category == OutcomeCategory.PAID:
+            counts["PAID"] += 1
+            continue
+
+        if o.category == OutcomeCategory.NOT_IN_835:
+            tf = timely_filing_decision(o.visit.date, as_of_date, override_filing)
+            if tf.rebillable:
+                pipeline_b_visits.append(o.visit)
+                counts[REBILLED_B] += 1
+                summary.append(SummaryRow(
+                    visit=o.visit,
+                    disposition=REBILLED_B,
+                    unpaid=o.unpaid,
+                    cas_reasons=(),
+                    file_path="pipeline-b/",
+                ))
+            else:
+                counts[EXCLUDED_TIMELY_FILING] += 1
+                summary.append(SummaryRow(
+                    visit=o.visit,
+                    disposition=EXCLUDED_TIMELY_FILING,
+                    unpaid=o.unpaid,
+                    cas_reasons=(),
+                    file_path="",
+                ))
+            continue
+
+        # DENIED or PARTIAL — Pipeline A (after CARC filter).
+        s = svc_lookup.get(
+            (o.visit.member_id, o.visit.procedure, o.visit.date)
+        )
+        reasons: tuple[str, ...] = s.cas_reasons if s else ()
+        decision = decide_carc(reasons)
+        if decision == CarcDecision.EXCLUDED:
+            counts[EXCLUDED_CARC] += 1
+            summary.append(SummaryRow(
+                visit=o.visit,
+                disposition=EXCLUDED_CARC,
+                unpaid=o.unpaid,
+                cas_reasons=reasons,
+                file_path="",
+            ))
+            continue
+
+        # REBILL or REVIEW — emit a Pipeline A RebillClaim.
+        # Each matched SVC carries its own original claim_id; preserve it
+        # per-claim so the freq-7 replacement is anchored to the right
+        # claim_submit_id. If the visit somehow matched nothing (shouldn't
+        # happen for DENIED/PARTIAL — those categories only arise from a
+        # match), mint a NEW-* fallback.
+        claim_id = (
+            s.claim_id if s
+            else f"NEW-{o.visit.member_id}-{o.visit.date.isoformat()}"
+        )
+        claim = RebillClaim(
+            claim_id=claim_id,
+            member_id=o.visit.member_id,
+            procedure=o.visit.procedure,
+            svc_date=o.visit.date,
+            charge=o.visit.billed,
+            frequency_code="7",
+            needs_review=(decision == CarcDecision.REVIEW),
+            original_carc_reasons=reasons,
+        )
+        pipeline_a_claims.append(claim)
+        counts[REBILLED_A] += 1
+        summary.append(SummaryRow(
+            visit=o.visit,
+            disposition=REBILLED_A,
+            unpaid=o.unpaid,
+            cas_reasons=reasons,
+            file_path="pipeline-a/",
+        ))
+
+    # 5) Emit Pipeline A files — placeholder; Task 13 wraps with the
+    #    serialize-then-Edifabric-validate step. Filename shape:
+    #    ``x-{claim_id}.837`` so the audit trail ties back to the
+    #    original claim_submit_id via the SummaryRow.cas_reasons chain.
+    a_dir = out_dir_p / "pipeline-a"
+    a_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"
+        p.touch()
+        a_files.append(p)
+
+    # 6) Build + emit Pipeline B batches — placeholder; Task 13 wraps
+    #    with the Edifabric gate. Filename shape mirrors the HCPF
+    #    member-week pattern: ``x-{member_id}-{iso_year}-W{iso_week:02d}.837``.
+    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] = []
+    for b in b_batches:
+        p = b_dir / f"x-{b.member_id}-{b.iso_year}-W{b.iso_week:02d}.837"
+        p.touch()
+        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,
+    )
\ No newline at end of file