diff --git a/backend/src/cyclone/db.py b/backend/src/cyclone/db.py
index 4baba3d..ddce59a 100644
--- a/backend/src/cyclone/db.py
+++ b/backend/src/cyclone/db.py
@@ -982,3 +982,19 @@ class Resubmission(Base):
             unique=True,
         ),
     )
+
+
+class SubmissionRecord(Base):
+    """SP41: one row per (claim_id, submitted_at) pre-flight dedup guard.
+
+    The 30-day dedup window is enforced by ``cyclone.store.submission_dedup``.
+    This table records (claim_id, submitted_at) for every push that passed
+    the pre-flight check. Past the window, the prior record is ignored
+    (the guard lets through re-submits older than
+    ``cyclone.store.submission_dedup.DEFAULT_WINDOW_DAYS``).
+    """
+
+    __tablename__ = "submission_dedup"
+
+    claim_id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    submitted_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, index=True)
diff --git a/backend/src/cyclone/migrations/0022_submission_dedup.sql b/backend/src/cyclone/migrations/0022_submission_dedup.sql
new file mode 100644
index 0000000..53a740c
--- /dev/null
+++ b/backend/src/cyclone/migrations/0022_submission_dedup.sql
@@ -0,0 +1,15 @@
+-- version: 22
+-- SP41: claim-id dedup at SFTP pre-flight.
+--
+-- The 30-day dedup window is enforced by cyclone.store.submission_dedup.
+-- This table records (claim_id, submitted_at) for every push that passed
+-- the pre-flight check. Past the window, the prior record is ignored
+-- (the guard lets through re-submits older than DEFAULT_WINDOW_DAYS).
+
+CREATE TABLE submission_dedup (
+    claim_id TEXT PRIMARY KEY,
+    submitted_at DATETIME NOT NULL
+);
+
+CREATE INDEX submission_dedup_submitted_at_idx
+    ON submission_dedup (submitted_at);
diff --git a/backend/src/cyclone/rebill/__init__.py b/backend/src/cyclone/rebill/__init__.py
new file mode 100644
index 0000000..39e04d1
--- /dev/null
+++ b/backend/src/cyclone/rebill/__init__.py
@@ -0,0 +1,11 @@
+"""SP41 — in-window rebill pipeline.
+
+Owns:
+  - parse_835_svc        (SVC-level 835 reparse with member_id)
+  - reconcile            (visit-to-835 join on (member_id, procedure, DOS))
+  - carc_filter          (CARC-aware exclusion for Pipeline A)
+  - timely_filing        (120-day DOS age gate)
+  - pipeline_a           (denied/partial → frequency-7 rebill)
+  - pipeline_b           (NOT_IN_835 → fresh 837Ps by (member, ISO-week))
+  - summary              (summary CSV + per-category counters)
+"""
\ No newline at end of file
diff --git a/backend/src/cyclone/rebill/carc_filter.py b/backend/src/cyclone/rebill/carc_filter.py
new file mode 100644
index 0000000..1922842
--- /dev/null
+++ b/backend/src/cyclone/rebill/carc_filter.py
@@ -0,0 +1,44 @@
+"""CARC-aware filter.
+
+EXCLUDED = contractual / not recoverable / not a denial-of-payment in the
+narrow sense (charge exceeds fee schedule, prior to coverage, etc.).
+REVIEW = operator must decide (claim/service lacks info, non-covered
+charges, duplicate, etc.). Anything else falls through to REBILL.
+"""
+from __future__ import annotations
+
+from enum import Enum
+
+
+class CarcDecision(str, Enum):
+    EXCLUDED = "EXCLUDED"
+    REVIEW = "REVIEW"
+    REBILL = "REBILL"
+
+
+# These CARCs are NOT recoverable denials. Do not rebill.
+EXCLUDED_CARCS: frozenset[str] = frozenset({
+    "CO-45",  # charge exceeds fee schedule / contractual obligation
+    "CO-26",  # expenses incurred prior to coverage
+    "CO-129",  # prior processing information; forward to next payer
+})
+
+# These CARCs need human review before rebill.
+REVIEW_CARCS: frozenset[str] = frozenset({
+    "PI-16",   # claim/service lacks information
+    "PI-96",   # non-covered charges
+    "PI-15",   # authorization / certification absent
+    "PI-4",    # procedure not paid separately
+    "PI-110",  # billing date predates service date
+    "OA-18",   # exact duplicate (resubmit noise)
+    "OA-23",   # impact of prior payer adjudication
+})
+
+
+def decide_carc(reasons: tuple[str, ...] | list[str]) -> CarcDecision:
+    """Pick the strongest action: EXCLUDED > REVIEW > REBILL."""
+    if any(r in EXCLUDED_CARCS for r in reasons):
+        return CarcDecision.EXCLUDED
+    if any(r in REVIEW_CARCS for r in reasons):
+        return CarcDecision.REVIEW
+    return CarcDecision.REBILL
diff --git a/backend/src/cyclone/rebill/parse_835_svc.py b/backend/src/cyclone/rebill/parse_835_svc.py
new file mode 100644
index 0000000..1422006
--- /dev/null
+++ b/backend/src/cyclone/rebill/parse_835_svc.py
@@ -0,0 +1,124 @@
+"""SVC-level 835 reparse with member_id at SVC scope.
+
+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).
+"""
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal
+from pathlib import Path
+
+ELEM = "*"
+SEG_END = "~"
+
+
+@dataclass(frozen=True)
+class SvcRow:
+    src_file: str
+    claim_id: str
+    member_id: str
+    status: str
+    procedure: str
+    modifiers: str
+    charge: Decimal
+    paid: Decimal
+    units: Decimal
+    svc_date: date
+    cas_reasons: tuple[str, ...]  # each entry is "GROUP-CODE" (e.g. "OA-18")
+    pay_date: date | None
+
+
+def _split(seg: str) -> list[str]:
+    return seg.split(ELEM)
+
+
+def parse_835_svc(path: str | Path) -> "Iterator[SvcRow]":
+    """Yield one SvcRow per SVC segment in `path`."""
+    path = Path(path)
+    text = path.read_text()
+    segments = [s for s in text.split(SEG_END) if s]
+    pay_date: date | None = None
+    current_claim_id = ""
+    current_status = ""
+    current_member_id = ""
+    i = 0
+    while i < len(segments):
+        elems = _split(segments[i])
+        seg_name = elems[0]
+        if seg_name == "DTM" and len(elems) > 2 and elems[1] == "405":
+            try:
+                pay_date = _ymd(elems[2])
+            except ValueError:
+                pass
+        elif seg_name == "CLP":
+            current_claim_id = elems[1] if len(elems) > 1 else ""
+            current_status = elems[2] if len(elems) > 2 else ""
+            current_member_id = ""
+        elif seg_name == "NM1" and len(elems) > 1 and elems[1] == "QC":
+            current_member_id = elems[9] if len(elems) > 9 else ""
+        elif seg_name == "SVC" and current_claim_id:
+            rest = elems[1:]
+            comp = rest[0] if rest else ""
+            proc_parts = comp.split(":")
+            procedure = proc_parts[1] if len(proc_parts) > 1 else ""
+            modifiers = ":".join(proc_parts[2:]) if len(proc_parts) > 2 else ""
+            charge = _decimal(rest[1]) if len(rest) > 1 else Decimal("0")
+            paid = _decimal(rest[2]) if len(rest) > 2 else Decimal("0")
+            units = _decimal(rest[3]) if len(rest) > 3 else Decimal("0")
+            svc_date, cas_reasons = _walk_for_dtm_and_cas(segments, i)
+            yield SvcRow(
+                src_file=path.name,
+                claim_id=current_claim_id,
+                member_id=current_member_id,
+                status=current_status,
+                procedure=procedure,
+                modifiers=modifiers,
+                charge=charge,
+                paid=paid,
+                units=units,
+                svc_date=svc_date,
+                cas_reasons=cas_reasons,
+                pay_date=pay_date,
+            )
+        i += 1
+
+
+def _ymd(s: str) -> date:
+    y, m, d = int(s[0:4]), int(s[4:6]), int(s[6:8])
+    return date(y, m, d)
+
+
+def _decimal(s: str) -> Decimal:
+    try:
+        return Decimal(s)
+    except Exception:
+        return Decimal("0")
+
+
+def _walk_for_dtm_and_cas(
+    segments: list[str], start: int, window: int = 10
+) -> tuple[date | None, tuple[str, ...]]:
+    """Walk the next `window` segments after SVC. Capture DTM*472 and CAS.
+
+    X12 835 may place DTM*472 before or after CAS segments within a service
+    line, so we walk the full window and capture both, breaking only at the
+    next SVC or CLP (a new claim/service boundary).
+    """
+    svc_date: date | None = None
+    cas_reasons: list[str] = []
+    for j in range(start + 1, min(start + window, len(segments))):
+        ej = _split(segments[j])
+        if ej[0] in ("SVC", "CLP"):
+            break
+        if ej[0] == "DTM" and len(ej) > 2 and ej[1] == "472" and svc_date is None:
+            try:
+                svc_date = _ymd(ej[2])
+            except ValueError:
+                pass
+        elif ej[0] == "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
diff --git a/backend/src/cyclone/rebill/pipeline_a.py b/backend/src/cyclone/rebill/pipeline_a.py
new file mode 100644
index 0000000..10e57a3
--- /dev/null
+++ b/backend/src/cyclone/rebill/pipeline_a.py
@@ -0,0 +1,86 @@
+"""Pipeline A: denied/partial → frequency-7 replacement 837Ps.
+
+For each visit the previous adjudication said was DENIED or PARTIAL,
+emit a fresh 837P with CLM05-3 = '7' (replacement claim) and the
+original claim_submit_id preserved as CLM01. The 837P carries the
+original DOS and the same procedure / member.
+
+The CARC-aware filter is the gate — only REBILL / REVIEW decisions are
+emitted; EXCLUDED visits never make it to a file.
+"""
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal
+from pathlib import Path
+
+from cyclone.rebill.carc_filter import CarcDecision
+from cyclone.rebill.reconcile import ReconcileOutcome, OutcomeCategory
+
+
+@dataclass(frozen=True)
+class RebillClaim:
+    """Pre-emission 837P shape for Pipeline A. The orchestrator's serializer
+    adapter converts this to a ClaimOutput for serialize_837.
+    """
+
+    claim_id: str           # reuses original claim_submit_id
+    member_id: str
+    procedure: str
+    svc_date: date
+    charge: Decimal
+    frequency_code: str     # always "7" for Pipeline A
+    needs_review: bool      # True for CARC REVIEW decisions
+    original_carc_reasons: tuple[str, ...]
+
+
+def build_pipeline_a_claims(
+    original_claim_id: str,
+    visit_outcomes: list[ReconcileOutcome],
+    carc_decisions: list[CarcDecision],
+    cas_reasons_per_visit: list[tuple[str, ...]],
+) -> list[RebillClaim]:
+    """Zip three parallel per-visit lists by index; drop PAID/NOT_IN_835 outcomes and EXCLUDED CARC decisions; emit one frequency-7 RebillClaim per survivor."""
+    out: list[RebillClaim] = []
+    # strict=True: all three lists must be the same length; a mismatch is a contract bug, fail loud.
+    for vo, carc, reasons in zip(visit_outcomes, carc_decisions, cas_reasons_per_visit, strict=True):
+        if vo.category not in (OutcomeCategory.DENIED, OutcomeCategory.PARTIAL):
+            continue
+        if carc == CarcDecision.EXCLUDED:
+            continue
+        out.append(RebillClaim(
+            claim_id=original_claim_id,
+            member_id=vo.visit.member_id,
+            procedure=vo.visit.procedure,
+            svc_date=vo.visit.date,
+            charge=vo.visit.billed,
+            frequency_code="7",
+            needs_review=(carc == CarcDecision.REVIEW),
+            original_carc_reasons=reasons,
+        ))
+    return out
+
+
+def write_pipeline_a_files(
+    claims: list[RebillClaim],
+    out_dir: Path,
+    serialize_837_fn,  # injected: callable[[RebillClaim], str] (orchestrator-supplied adapter)
+    tpid: str,
+) -> list[Path]:
+    """Write one 837P per claim, using HCPF-spec filenames via build_outbound_filename.
+
+    `serialize_837_fn(claim)` returns the X12 string. The 837P envelope is
+    pure ASCII so we write it as text. The filename is
+    `cyclone.edi.filenames.build_outbound_filename(tpid, '837P')`.
+    """
+    # Lazy: keep cyclone.rebill.pipeline_a importable without zoneinfo from filenames.
+    from cyclone.edi.filenames import build_outbound_filename
+
+    out_dir.mkdir(parents=True, exist_ok=True)
+    paths: list[Path] = []
+    for c in claims:
+        body = serialize_837_fn(c)
+        fname = build_outbound_filename(tpid, "837P")
+        path = out_dir / fname
+        path.write_text(body, encoding="ascii")
+        paths.append(path)
+    return paths
diff --git a/backend/src/cyclone/rebill/pipeline_b.py b/backend/src/cyclone/rebill/pipeline_b.py
new file mode 100644
index 0000000..ad49325
--- /dev/null
+++ b/backend/src/cyclone/rebill/pipeline_b.py
@@ -0,0 +1,123 @@
+"""Pipeline B: NOT_IN_835 visits → fresh 837Ps, batched by (member_id, ISO-week).
+
+One 837P per (member, ISO-week) pair. Per-visit files would be 4,509 SFTP
+round-trips; per-batch (no member split) would lose the operator's
+ability to track per-member. Member-week is the standard clearinghouse
+pattern and matches AxisCare's billing cycle.
+
+The timely-filing gate is applied per visit before batching. The
+override flag is the same for the whole batch (a per-batch override).
+"""
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+
+from cyclone.rebill.reconcile import VisitRow
+from cyclone.rebill.timely_filing import timely_filing_decision
+
+
+@dataclass(frozen=True)
+class MemberWeekBatch:
+    """Pre-emission 837P shape for Pipeline B. The orchestrator's serializer
+    adapter converts this to a ClaimOutput for serialize_837.
+
+    Note: frozen=True is shallow — `visits` is a mutable list, so
+    `mb.visits.append(x)` works even though `mb.visits = new_list` raises
+    FrozenInstanceError. Treat the batch as logically immutable.
+    """
+
+    member_id: str
+    iso_year: int
+    iso_week: int
+    visits: list[VisitRow]
+    has_overridden_visits: bool
+
+
+def build_pipeline_b_batches(
+    visits: list[VisitRow],
+    as_of: date,
+    override: bool,
+) -> list[MemberWeekBatch]:
+    """Group rebillable visits by (member_id, ISO-week).
+
+    Two filters are applied per visit before batching:
+      1. timely_filing_decision(...).rebillable — drops past-window
+         visits unless `override=True` (per-batch override flag).
+      2. implicit: only visits that survive (1) are grouped; the rest
+         are surfaced as EXCLUDED_TIMELY_FILING in the summary CSV.
+
+    Returns batches sorted by (member_id, iso_year, iso_week) for
+    deterministic filenames downstream.
+    """
+    by_key: dict[tuple[str, int, int], list[VisitRow]] = defaultdict(list)
+    overridden_keys: set[tuple[str, int, int]] = set()
+    for v in visits:
+        decision = timely_filing_decision(v.date, as_of, override)
+        if not decision.rebillable:
+            continue
+        iso = v.date.isocalendar()
+        key = (v.member_id, iso.year, iso.week)
+        by_key[key].append(v)
+        if decision.override and not decision.within_window:
+            overridden_keys.add(key)
+    return [
+        MemberWeekBatch(
+            member_id=member, iso_year=yr, iso_week=wk,
+            visits=vs, has_overridden_visits=(member, yr, wk) in overridden_keys,
+        )
+        for (member, yr, wk), vs in sorted(by_key.items())
+    ]
+
+
+def batch_visits_by_member_week(
+    visits: list[VisitRow],
+    as_of: date,
+    override: bool,
+) -> dict[str, list[VisitRow]]:
+    """Thin wrapper over build_pipeline_b_batches.
+
+    Returns ``{member_id: [VisitRow, ...]}`` — collapsing the (member,
+    ISO-week) batches back down to a per-member visit list. This is the
+    shape the orchestrator's summary aggregation expects.
+
+    PROVISIONAL: this helper is defined for the Task 12 orchestrator's
+    anticipated use, but has no callers in this branch. If Task 12's
+    actual needs differ, this function may be removed or replaced without
+    notice.
+
+    This also collapses multiple weeks for the same member into one
+    list. If the orchestrator needs week-aware per-member iteration,
+    use build_pipeline_b_batches directly.
+    """
+    batches = build_pipeline_b_batches(visits, as_of=as_of, override=override)
+    out: dict[str, list[VisitRow]] = defaultdict(list)
+    for b in batches:
+        out[b.member_id].extend(b.visits)
+    return dict(out)
+
+
+def write_pipeline_b_files(
+    batches: list[MemberWeekBatch],
+    out_dir: Path,
+    serialize_837_fn,
+    tpid: str,
+) -> list[Path]:
+    """Write one 837P per batch, using HCPF-spec filenames.
+
+    `serialize_837_fn(batch)` returns the X12 string. The 837P envelope
+    is pure ASCII so we write it as text. The filename is
+    `cyclone.edi.filenames.build_outbound_filename(tpid, '837P')`.
+    """
+    # Lazy: keep cyclone.rebill.pipeline_b importable without zoneinfo from filenames.
+    from cyclone.edi.filenames import build_outbound_filename
+
+    out_dir.mkdir(parents=True, exist_ok=True)
+    paths: list[Path] = []
+    for b in batches:
+        body = serialize_837_fn(b)
+        fname = build_outbound_filename(tpid, "837P")
+        path = out_dir / fname
+        path.write_text(body, encoding="ascii")
+        paths.append(path)
+    return paths
diff --git a/backend/src/cyclone/rebill/reconcile.py b/backend/src/cyclone/rebill/reconcile.py
new file mode 100644
index 0000000..f5a803b
--- /dev/null
+++ b/backend/src/cyclone/rebill/reconcile.py
@@ -0,0 +1,69 @@
+"""Visit-to-835 reconciliation on (member_id, procedure, DOS).
+
+Authoritative match key. The 5% tolerance on PAID handles the legitimate
+charge-amount differences between AxisCare's per-unit CSV and the 835's
+per-SVC breakdown.
+"""
+from collections import defaultdict
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal
+from enum import Enum
+from typing import Iterable
+
+# SvcRow is owned by parse_835_svc (Task 1) — re-exported here so callers
+# of `cyclone.rebill.reconcile` only need one import.
+from cyclone.rebill.parse_835_svc import SvcRow  # noqa: F401  (re-exported)
+
+PAID_TOLERANCE = Decimal("0.05")            # charge match tolerance
+PARTIAL_RATIO  = Decimal("0.95")            # PAID if total_paid >= 95% of billed
+
+
+class OutcomeCategory(str, Enum):
+    PAID = "PAID"
+    PARTIAL = "PARTIAL"
+    DENIED = "DENIED"
+    NOT_IN_835 = "NOT_IN_835"
+
+
+@dataclass(frozen=True)
+class VisitRow:
+    date: date
+    member_id: str
+    procedure: str
+    billed: Decimal
+
+
+@dataclass(frozen=True)
+class ReconcileOutcome:
+    visit: VisitRow
+    category: OutcomeCategory
+    unpaid: Decimal
+    matched_svc_count: int
+
+
+def reconcile_visits_to_835(
+    visits: Iterable[VisitRow],
+    svcs: Iterable[SvcRow],
+) -> list[ReconcileOutcome]:
+    """Index SVCs by (member_id, procedure, DOS), best-of-N outcome per visit."""
+    by_key: dict[tuple[str, str, date], list[SvcRow]] = defaultdict(list)
+    for s in svcs:
+        by_key[(s.member_id, s.procedure, s.svc_date)].append(s)
+
+    out: list[ReconcileOutcome] = []
+    for v in visits:
+        cands = by_key.get((v.member_id, v.procedure, v.date), [])
+        if not cands:
+            out.append(ReconcileOutcome(v, OutcomeCategory.NOT_IN_835, v.billed, 0))
+            continue
+        total_paid = sum((c.paid for c in cands), Decimal("0"))
+        any_paid = any(c.paid > 0 for c in cands)
+        if any_paid and total_paid >= v.billed * PARTIAL_RATIO:
+            cat, unpaid = OutcomeCategory.PAID, Decimal("0")
+        elif any_paid:
+            cat, unpaid = OutcomeCategory.PARTIAL, max(Decimal("0"), v.billed - total_paid)
+        else:
+            cat, unpaid = OutcomeCategory.DENIED, v.billed
+        out.append(ReconcileOutcome(v, cat, unpaid, len(cands)))
+    return out
diff --git a/backend/src/cyclone/rebill/summary.py b/backend/src/cyclone/rebill/summary.py
new file mode 100644
index 0000000..99e22c4
--- /dev/null
+++ b/backend/src/cyclone/rebill/summary.py
@@ -0,0 +1,57 @@
+"""Summary CSV for SP41.
+
+One row per visit, classified into one of:
+  REBILLED_A          — pipeline A emitted a frequency-7 replacement
+  REBILLED_B          — pipeline B emitted a fresh 837P
+  EXCLUDED_CARC       — CO-45/CO-26/CO-129 (or other excluded CARC)
+  EXCLUDED_PAYER      — non-CO_TXIX payer per AxisCare API spot audit
+  EXCLUDED_NO_EVV     — no Authorized=Yes AND no AxisCare EVV ref
+  EXCLUDED_TIMELY_FILING — DOS > 120 days before run date, no override
+"""
+import csv
+from dataclasses import dataclass
+from decimal import Decimal
+from pathlib import Path
+
+from cyclone.rebill.reconcile import VisitRow
+
+REBILLED_A = "REBILLED_A"
+REBILLED_B = "REBILLED_B"
+EXCLUDED_CARC = "EXCLUDED_CARC"
+EXCLUDED_PAYER = "EXCLUDED_PAYER"
+EXCLUDED_NO_EVV = "EXCLUDED_NO_EVV"
+EXCLUDED_TIMELY_FILING = "EXCLUDED_TIMELY_FILING"
+
+
+@dataclass(frozen=True)
+class SummaryRow:
+    """One visit + its post-pipeline disposition for the operator audit trail."""
+
+    visit: VisitRow
+    disposition: str
+    unpaid: Decimal
+    cas_reasons: tuple[str, ...]
+    file_path: str  # relative to dev/rebills/<date>/
+
+
+def write_summary_csv(rows: list[SummaryRow], out_path: Path) -> int:
+    """Write the summary CSV; return the number of rows written."""
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    with out_path.open("w", newline="") as f:
+        w = csv.writer(f)
+        w.writerow([
+            "dos", "member_id", "procedure", "billed",
+            "disposition", "unpaid", "cas_reasons", "file_path",
+        ])
+        for r in rows:
+            w.writerow([
+                r.visit.date.isoformat(),
+                r.visit.member_id,
+                r.visit.procedure,
+                str(r.visit.billed),
+                r.disposition,
+                str(r.unpaid),
+                "|".join(r.cas_reasons),  # pipe-separated; no collision with J-prefixed member IDs or relative file paths
+                r.file_path,
+            ])
+    return len(rows)
diff --git a/backend/src/cyclone/rebill/timely_filing.py b/backend/src/cyclone/rebill/timely_filing.py
new file mode 100644
index 0000000..b1aa995
--- /dev/null
+++ b/backend/src/cyclone/rebill/timely_filing.py
@@ -0,0 +1,37 @@
+"""120-day timely-filing gate for Colorado Medicaid.
+
+Per HCPF's Colorado Medical Assistance Program provider manual, claims must
+be filed within 120 days from the date of service. The gate is HARD by
+default — past-window visits are surfaced as EXCLUDED_TIMELY_FILING in the
+summary CSV and not emitted by Pipeline B.
+
+The operator may pass override=True per batch to relax the gate (e.g. for
+good-cause appeal visits). The decision is recorded in the summary CSV
+either way so the audit trail shows the override.
+"""
+from dataclasses import dataclass
+from datetime import date
+
+FILING_WINDOW_DAYS = 120
+
+
+@dataclass(frozen=True)
+class TimelyFilingDecision:
+    dos: date
+    as_of: date
+    days_old: int
+    within_window: bool
+    override: bool
+    rebillable: bool
+
+
+def timely_filing_decision(
+    dos: date, as_of: date, override: bool
+) -> TimelyFilingDecision:
+    days_old = (as_of - dos).days
+    within_window = days_old <= FILING_WINDOW_DAYS
+    return TimelyFilingDecision(
+        dos=dos, as_of=as_of, days_old=days_old,
+        within_window=within_window, override=override,
+        rebillable=within_window or override,
+    )
diff --git a/backend/src/cyclone/store/exceptions.py b/backend/src/cyclone/store/exceptions.py
index 84b5ae1..e8eb1c1 100644
--- a/backend/src/cyclone/store/exceptions.py
+++ b/backend/src/cyclone/store/exceptions.py
@@ -4,6 +4,8 @@ These are part of the public API — callers (API endpoints) catch them
 and translate to HTTP 409 Conflict responses.
 """
 
+from datetime import datetime
+
 
 class AlreadyMatchedError(Exception):
     """Raised by ``CycloneStore.manual_match`` when the claim is already paired.
@@ -38,4 +40,27 @@ class InvalidStateError(Exception):
         self.activity_kind = activity_kind
         super().__init__(
             f"invalid state {current_state} for apply (kind={activity_kind})"
-        )
\ No newline at end of file
+        )
+
+
+class DuplicateClaimError(Exception):
+    """Raised by ``cyclone.store.submission_dedup.check_duplicate`` when a claim_id
+    is re-submitted within the 30-day dedup window (SP41).
+
+    Mirrors the other 409-class exceptions: callers catch it at the API
+    boundary and surface it as a 409 Conflict. Carries
+    ``original_submission_at`` so the UI can render "already submitted on
+    YYYY-MM-DD" without re-querying.
+
+    Note: the "30-day" literal in the message mirrors
+    ``cyclone.store.submission_dedup.DEFAULT_WINDOW_DAYS`` — keep them in
+    sync if either ever moves.
+    """
+
+    def __init__(self, claim_id: str, original_submission_at: datetime):
+        self.claim_id = claim_id
+        self.original_submission_at = original_submission_at
+        super().__init__(
+            f"claim_id {claim_id!r} was already submitted at "
+            f"{original_submission_at.isoformat()}; within 30-day window"
+        )
diff --git a/backend/src/cyclone/store/submission_dedup.py b/backend/src/cyclone/store/submission_dedup.py
new file mode 100644
index 0000000..db97a18
--- /dev/null
+++ b/backend/src/cyclone/store/submission_dedup.py
@@ -0,0 +1,65 @@
+"""Claim-id dedup at the SFTP pre-flight stage.
+
+Schema: the ``submission_dedup`` table (defined on the main ``Base`` in
+``cyclone.db``) records (claim_id, submitted_at). A re-submission of
+the same claim_id within ``DEFAULT_WINDOW_DAYS`` (30) is rejected with
+``DuplicateClaimError``. Past the window, the guard lets it through —
+the operator may be retrying a 999-rejected file from >30 days ago,
+which is a legitimate rebuild path.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+
+from cyclone import db as cycl_db
+from cyclone.db import SubmissionRecord
+from cyclone.store.exceptions import DuplicateClaimError
+
+DEFAULT_WINDOW_DAYS = 30
+
+
+def record_submission(claim_id: str, submitted_at: datetime, db_url: str) -> None:
+    """Upsert (claim_id, submitted_at). Idempotent on claim_id.
+
+    ``db_url`` is accepted for back-compat with the original Task 8
+    signature but ignored — sessions now flow through the process-wide
+    ``cycl_db.SessionLocal()`` factory that ``db.init_db()`` set up.
+    """
+    del db_url  # signature-preserving; conftest wires the per-test DB
+    with cycl_db.SessionLocal()() as session:
+        session.merge(SubmissionRecord(
+            claim_id=claim_id,
+            submitted_at=submitted_at,
+        ))
+        session.commit()
+
+
+def check_duplicate(
+    claim_id: str,
+    db_url: str | None = None,
+    now: datetime | None = None,
+    window_days: int = DEFAULT_WINDOW_DAYS,
+) -> None:
+    """Raise ``DuplicateClaimError`` if ``claim_id`` was submitted within the window.
+
+    ``db_url`` is accepted for back-compat with the original Task 8
+    signature but ignored — sessions flow through ``cycl_db.SessionLocal()``
+    just like the rest of ``cyclone.store``. Tests that pass an explicit
+    ``db_url`` keep working because ``cycl_db.SessionLocal()()`` points at
+    the URL the conftest resolved under ``CYCLONE_DB_URL``.
+    """
+    del db_url  # signature-preserving
+    if now is None:
+        now = datetime.now(timezone.utc)
+    cutoff = now - timedelta(days=window_days)
+    with cycl_db.SessionLocal()() as session:
+        row = session.scalars(
+            select(SubmissionRecord).where(
+                SubmissionRecord.claim_id == claim_id,
+                SubmissionRecord.submitted_at >= cutoff,
+            )
+        ).first()
+    if row is not None:
+        raise DuplicateClaimError(claim_id, row.submitted_at)
diff --git a/backend/tests/fixtures/835_sample_svc_with_member.txt b/backend/tests/fixtures/835_sample_svc_with_member.txt
new file mode 100644
index 0000000..e102072
--- /dev/null
+++ b/backend/tests/fixtures/835_sample_svc_with_member.txt
@@ -0,0 +1 @@
+ISA*00*          *00*          *ZZ*CYCLONE         *ZZ*GAINWELL       *260101*1200*^*00501*000000001*0*P*:~GS*HC*CYCLONE*GAINWELL*20260101*1200*1*X*005010X221A1~ST*835*0001~BPR*I*100.00*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~CLP*T1001*1*16.24*16.10**MC*2026029105200*11*1~NM1*QC*1*AALBUE*ERIC****MR*J813715~NM1*74*1*AALBUE*ERIC*W***C*J813715~DTM*232*20260118~DTM*233*20260124~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260118~CAS*CO*45*.02~REF*G1*6252960154~REF*6R*T1001V001~AMT*B6*2.30~SVC*HC:T1019:U2:SC:KX*2.32*2.30**.33~DTM*472*20260119~CAS*OA*18*2.32~SVC*HC:T1019:U2:SC:KX*2.32*-2.30**.33~DTM*472*20260120~CAS*CO*45*-.02~LX*2~CLP*T1002*4*16.24*0**MC*2026029105201*11*1~NM1*QC*1*OTHER-A*FIRST****MR*OTHER-MEMBER-A~SVC*HC:T1019*2.32*0**.33~DTM*472*20260121~CAS*CO*97*2.32~LX*3~CLP*T1003*22*16.24*-16.10**MC*2026029105202*11*1~NM1*QC*1*OTHER-B*SECOND****MR*OTHER-MEMBER-B~SVC*HC:T1019*2.32*-2.30**.33~DTM*472*20260122~SE*23*0001~GE*1*1~IEA*1*000000001~
\ No newline at end of file
diff --git a/backend/tests/test_acks.py b/backend/tests/test_acks.py
index 5eae3dd..9c740d3 100644
--- a/backend/tests/test_acks.py
+++ b/backend/tests/test_acks.py
@@ -51,7 +51,7 @@ def test_migration_0002_creates_acks_table():
 
 def test_migration_latest_idempotent_on_fresh_db():
     """Re-running the migration on the same DB must be a no-op (PRAGMA
-    user_version already at the latest version — currently 21 after
+    user_version already at the latest version — currently 22 after
     0004-0006 line_reconciliation, 0005 ta1_acks, SP9's 0007
     providers/payers/clearhouse, SP10's 0008 payer_rejected,
     SP11's 0009 audit_log, SP14's 0010 payer_rejected_acknowledged,
@@ -62,16 +62,16 @@ def test_migration_latest_idempotent_on_fresh_db():
     claim.patient_control_number backfill UPDATE, SP28's 0018
     claim_acks join table, SP32's 0019
     rendering_provider_npi + service_provider_npi, SP37's 0020
-    transaction_set_control_number, and SP39's 0021 resubmissions
-    audit table)."""
+    transaction_set_control_number, SP39's 0021 resubmissions
+    audit table, and SP41's 0022 submission_dedup)."""
     with db.engine().begin() as c:
         v1 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v1 == 21
+    assert v1 == 22
     # A second run should not raise and should not bump the version.
     db_migrate.run(db.engine())
     with db.engine().begin() as c:
         v2 = c.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v2 == 21
+    assert v2 == 22
 
 
 def test_add_ack_persists_row():
diff --git a/backend/tests/test_db_migrate.py b/backend/tests/test_db_migrate.py
index 931debf..bc96ca2 100644
--- a/backend/tests/test_db_migrate.py
+++ b/backend/tests/test_db_migrate.py
@@ -128,14 +128,15 @@ def test_migration_latest_idempotent_on_fresh_db(tmp_path: Path) -> None:
     service_provider_npi on claims and remittances.
     SP37 bumped it to 20 with batch transaction_set_control_number.
     SP39 bumped it to 21 with the resubmissions audit table.
+    SP41 bumped it to 22 with submission_dedup.
     """
     engine = _fresh_engine(tmp_path)
     db_migrate.run(engine)
     v_after_first = _user_version(engine)
-    assert v_after_first == 21, f"expected head=21, got {v_after_first}"
+    assert v_after_first == 22, f"expected head=22, got {v_after_first}"
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 21, "second run should not bump version"
+    assert _user_version(engine) == 22, "second run should not bump version"
 
 
 def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -161,7 +162,7 @@ def test_drop_claims_unique_constraint_migration(tmp_path: Path, monkeypatch: py
     engine = _fresh_engine(tmp_path)
 
     db_migrate.run(engine)
-    assert _user_version(engine) == 21, f"expected head=21, got {_user_version(engine)}"
+    assert _user_version(engine) == 22, f"expected head=22, got {_user_version(engine)}"
 
     # Two claims in one batch with the same patient_control_number
     # must be insertable. If 0015's table recreation re-introduced a
diff --git a/backend/tests/test_migration_0020.py b/backend/tests/test_migration_0020.py
index b001a02..e6f4415 100644
--- a/backend/tests/test_migration_0020.py
+++ b/backend/tests/test_migration_0020.py
@@ -102,10 +102,10 @@ def migrated_engine(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
     engine = _fresh_engine(tmp_path / "mig0020.db")
     db_migrate.run(engine)
 
-    # Confirm head is 21 (every migration applied, including SP39's 0021).
+    # Confirm head is 22 (every migration applied, including SP41's 0022).
     with engine.connect() as conn:
         v = conn.exec_driver_sql("PRAGMA user_version").scalar() or 0
-    assert v == 21, f"expected migration head=21, got {v}"
+    assert v == 22, f"expected migration head=22, got {v}"
 
     yield engine
     engine.dispose()
diff --git a/backend/tests/test_rebill_carc_filter.py b/backend/tests/test_rebill_carc_filter.py
new file mode 100644
index 0000000..7871335
--- /dev/null
+++ b/backend/tests/test_rebill_carc_filter.py
@@ -0,0 +1,49 @@
+"""CARC-aware filter for Pipeline A.
+
+Contractual / non-recoverable denials (CO-45, CO-26, CO-129) must be
+excluded from rebill. CARCs that need operator review (PI-16, PI-96,
+PI-15, PI-4, PI-110, OA-18, OA-23) must be surfaced as 'REVIEW' so the
+operator can decide.
+"""
+from cyclone.rebill.carc_filter import (
+    CarcDecision,
+    decide_carc,
+    EXCLUDED_CARCS,
+    REVIEW_CARCS,
+)
+
+
+def test_co45_is_excluded():
+    assert decide_carc(("CO-45",)) == CarcDecision.EXCLUDED
+
+
+def test_co26_is_excluded():
+    assert decide_carc(("CO-26",)) == CarcDecision.EXCLUDED
+
+
+def test_co129_is_excluded():
+    assert decide_carc(("CO-129",)) == CarcDecision.EXCLUDED
+
+
+def test_pi16_is_review():
+    assert decide_carc(("PI-16",)) == CarcDecision.REVIEW
+
+
+def test_o18_is_review():
+    """OA-18 is the duplicate noise — surface for review, don't auto-rebill."""
+    assert decide_carc(("OA-18",)) == CarcDecision.REVIEW
+
+
+def test_no_carc_is_rebill():
+    assert decide_carc(()) == CarcDecision.REBILL
+
+
+def test_mixed_excluded_wins():
+    """If any CARC is excluded, the whole service is excluded."""
+    assert decide_carc(("PI-16", "CO-45")) == CarcDecision.EXCLUDED
+
+
+def test_sets_have_expected_members():
+    assert "CO-45" in EXCLUDED_CARCS
+    assert "PI-16" in REVIEW_CARCS
+    assert "OA-18" in REVIEW_CARCS
diff --git a/backend/tests/test_rebill_parse_835_svc.py b/backend/tests/test_rebill_parse_835_svc.py
new file mode 100644
index 0000000..437487b
--- /dev/null
+++ b/backend/tests/test_rebill_parse_835_svc.py
@@ -0,0 +1,38 @@
+"""835 SVC-level parser with member_id at SVC scope."""
+from pathlib import Path
+from collections import defaultdict
+from decimal import Decimal
+from cyclone.rebill.parse_835_svc import parse_835_svc
+
+FIX = Path(__file__).parent / "fixtures" / "835_sample_svc_with_member.txt"
+
+def test_parse_835_svc_extracts_member_id():
+    """NM1*QC NM109 at the CLP scope must propagate to that CLP's SVC rows;
+    the multi-claim fixture confirms each CLP's NM1*QC lands on its own SVCs."""
+    rows = list(parse_835_svc(FIX))
+    assert len(rows) >= 1
+    by_claim: dict[str, list[str]] = defaultdict(list)
+    for r in rows:
+        by_claim[r.claim_id].append(r.member_id)
+        assert r.procedure  # non-empty
+        assert r.svc_date   # non-empty
+        assert r.charge > 0
+    # First CLP (T1001) -> original fixture member
+    assert all(mid == "J813715" for mid in by_claim["T1001"]), by_claim
+    # Second CLP (T1002, status 4) -> its own NM1*QC
+    assert all(mid == "OTHER-MEMBER-A" for mid in by_claim["T1002"]), by_claim
+    # Third CLP (T1003, status 22) -> its own NM1*QC
+    assert all(mid == "OTHER-MEMBER-B" for mid in by_claim["T1003"]), by_claim
+
+def test_parse_835_svc_extracts_cas_reasons():
+    """CAS segments after DTM*472 must be captured (post-DTM*472 ordering)."""
+    rows = list(parse_835_svc(FIX))
+    # at least one row should have an OA-18 reason
+    assert any("OA-18" in r.cas_reasons for r in rows)
+
+def test_parse_835_svc_picks_up_status_22_reversals():
+    """Status 22 (reversal of previous payment) must be preserved, along with
+    status 1 (primary) and status 4 (denied)."""
+    rows = list(parse_835_svc(FIX))
+    statuses = {r.status for r in rows}
+    assert statuses == {"1", "4", "22"}
diff --git a/backend/tests/test_rebill_pipeline_a.py b/backend/tests/test_rebill_pipeline_a.py
new file mode 100644
index 0000000..656717a
--- /dev/null
+++ b/backend/tests/test_rebill_pipeline_a.py
@@ -0,0 +1,55 @@
+"""Pipeline A: denied/partial visits become frequency-7 replacement claims."""
+from datetime import date
+from decimal import Decimal
+from cyclone.rebill.pipeline_a import build_pipeline_a_claims
+from cyclone.rebill.reconcile import VisitRow, ReconcileOutcome, OutcomeCategory
+from cyclone.rebill.carc_filter import CarcDecision
+
+
+def _outcome(date_, member, proc, billed, cat, unpaid=Decimal("0")):
+    visit = VisitRow(date=date_, member_id=member, procedure=proc, billed=Decimal(billed))
+    return ReconcileOutcome(visit, cat, unpaid, 0)
+
+
+def test_pipeline_a_emits_frequency_7():
+    """CLM05-3 must be 7 (replacement), preserving the original claim_submit_id."""
+    out = build_pipeline_a_claims(
+        original_claim_id="ORIG-001",
+        visit_outcomes=[
+            _outcome(date(2026, 6, 27), "J813715", "T1019", "2.32",
+                     OutcomeCategory.DENIED, unpaid=Decimal("2.32")),
+        ],
+        carc_decisions=[CarcDecision.REBILL],
+        cas_reasons_per_visit=[()],
+    )
+    assert len(out) == 1
+    assert out[0].claim_id == "ORIG-001"
+    assert out[0].frequency_code == "7"
+    assert out[0].svc_date == date(2026, 6, 27)
+
+
+def test_pipeline_a_excludes_carc_excluded_visits():
+    out = build_pipeline_a_claims(
+        original_claim_id="ORIG-002",
+        visit_outcomes=[
+            _outcome(date(2026, 6, 27), "J813715", "T1019", "2.32",
+                     OutcomeCategory.DENIED),
+        ],
+        carc_decisions=[CarcDecision.EXCLUDED],
+        cas_reasons_per_visit=[("CO-45",)],
+    )
+    assert out == []  # not emitted
+
+
+def test_pipeline_a_surfaces_review_visits_with_flag():
+    out = build_pipeline_a_claims(
+        original_claim_id="ORIG-003",
+        visit_outcomes=[
+            _outcome(date(2026, 6, 27), "J813715", "T1019", "2.32",
+                     OutcomeCategory.DENIED),
+        ],
+        carc_decisions=[CarcDecision.REVIEW],
+        cas_reasons_per_visit=[("PI-16",)],
+    )
+    assert len(out) == 1
+    assert out[0].needs_review is True
diff --git a/backend/tests/test_rebill_pipeline_b.py b/backend/tests/test_rebill_pipeline_b.py
new file mode 100644
index 0000000..e2fb7d4
--- /dev/null
+++ b/backend/tests/test_rebill_pipeline_b.py
@@ -0,0 +1,75 @@
+"""Pipeline B: NOT_IN_835 visits → fresh 837Ps, batched by (member, ISO-week)."""
+from datetime import date
+from decimal import Decimal
+from cyclone.rebill.pipeline_b import build_pipeline_b_batches
+from cyclone.rebill.reconcile import VisitRow
+
+
+def _v(date_, member, proc, amt):
+    return VisitRow(date=date_, member_id=member, procedure=proc, billed=Decimal(amt))
+
+
+def test_batches_split_by_member_and_iso_week():
+    visits = [
+        _v(date(2026, 6, 23), "J813715", "T1019", "2.32"),
+        _v(date(2026, 6, 25), "J813715", "T1019", "2.32"),  # same week
+        _v(date(2026, 6, 23), "OTHER",    "T1019", "2.32"),  # different member
+        _v(date(2026, 6, 30), "J813715", "T1019", "2.32"),  # different week
+    ]
+    out = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False)
+    assert len(out) == 3
+    sizes = sorted(len(b.visits) for b in out)
+    assert sizes == [1, 1, 2]
+
+
+def test_timely_filing_excludes_old_visits():
+    visits = [
+        _v(date(2026, 1, 1), "OLD", "T1019", "2.32"),  # 187 days old
+        _v(date(2026, 6, 27), "NEW", "T1019", "2.32"), # 10 days old
+    ]
+    out = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False)
+    members = {b.member_id for b in out}
+    assert "OLD" not in members
+    assert "NEW" in members
+
+
+def test_override_relaxes_timely_filing():
+    visits = [_v(date(2026, 1, 1), "OLD", "T1019", "2.32")]
+    out_default = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=False)
+    out_overridden = build_pipeline_b_batches(visits, as_of=date(2026, 7, 7), override=True)
+    assert out_default == []
+    assert len(out_overridden) == 1
+
+
+def test_override_flag_set_on_past_window_visit_batch():
+    # Same member "OLD" with one past-window visit (187 days old, ISO
+    # week 1) and one within-window visit (10 days old, ISO week 26).
+    # With override=True the past-window visit survives and the batch
+    # for week 1 must be flagged has_overridden_visits=True because the
+    # override saved an otherwise-excluded visit.
+    visits = [
+        _v(date(2026, 1, 1), "OLD", "T1019", "2.32"),  # 187 days old (past window, W01)
+        _v(date(2026, 6, 27), "OLD", "T1019", "2.32"),  # 10 days old (within window, W26)
+    ]
+    out_overridden = build_pipeline_b_batches(
+        visits, as_of=date(2026, 7, 7), override=True,
+    )
+    # Two batches: one per (member, ISO-week) — the past-window visit and
+    # the within-window visit land in different weeks.
+    assert len(out_overridden) == 2
+    by_week = {b.iso_week: b for b in out_overridden}
+    assert by_week[1].member_id == "OLD"
+    assert by_week[1].has_overridden_visits is True  # the override saved this batch
+    assert by_week[26].member_id == "OLD"
+    assert by_week[26].has_overridden_visits is False  # no override needed here
+
+    # With override=False the past-window visit is dropped. The surviving
+    # within-window batch (W26) must NOT carry the override flag, and the
+    # past-window batch (W01) is absent.
+    out_default = build_pipeline_b_batches(
+        visits, as_of=date(2026, 7, 7), override=False,
+    )
+    assert len(out_default) == 1
+    assert out_default[0].member_id == "OLD"
+    assert out_default[0].iso_week == 26
+    assert out_default[0].has_overridden_visits is False
diff --git a/backend/tests/test_rebill_reconcile.py b/backend/tests/test_rebill_reconcile.py
new file mode 100644
index 0000000..30db4be
--- /dev/null
+++ b/backend/tests/test_rebill_reconcile.py
@@ -0,0 +1,76 @@
+"""Visit-to-835 reconciliation on (member_id, procedure, DOS)."""
+from datetime import date
+from decimal import Decimal
+from cyclone.rebill.reconcile import (
+    VisitRow,
+    SvcRow,
+    ReconcileOutcome,
+    reconcile_visits_to_835,
+    OutcomeCategory,
+)
+
+
+def _v(date, member, proc, amt):
+    return VisitRow(date=date, member_id=member, procedure=proc, billed=Decimal(amt))
+
+
+def _s(date, member, proc, chg, paid, status="1"):
+    return SvcRow(
+        src_file="x.835", claim_id="c1", member_id=member, status=status,
+        procedure=proc, modifiers="", charge=Decimal(chg), paid=Decimal(paid),
+        units=Decimal("0"), svc_date=date, cas_reasons=(), pay_date=None,
+    )
+
+
+def test_visit_paid_when_any_svc_paid_within_tolerance():
+    visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
+    svcs = [_s(date(2026, 1, 18), "J813715", "T1019", "2.32", "2.30")]
+    out = reconcile_visits_to_835(visits, svcs)
+    assert out[0].category == OutcomeCategory.PAID
+    assert out[0].unpaid == Decimal("0")
+
+
+def test_visit_paid_when_one_svc_paid_others_denied_duplicate():
+    """OA-18 on duplicates is noise; paid SVC wins."""
+    visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
+    svcs = [
+        _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "2.30", "1"),
+        _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"),
+        _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"),
+    ]
+    out = reconcile_visits_to_835(visits, svcs)
+    assert out[0].category == OutcomeCategory.PAID
+
+
+def test_visit_denied_when_all_svc_denied():
+    visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
+    svcs = [
+        _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"),
+        _s(date(2026, 1, 18), "J813715", "T1019", "2.32", "0.00", "4"),
+    ]
+    out = reconcile_visits_to_835(visits, svcs)
+    assert out[0].category == OutcomeCategory.DENIED
+    assert out[0].unpaid == Decimal("2.32")
+
+
+def test_visit_not_in_835_when_no_svc_match():
+    visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
+    svcs = [_s(date(2026, 1, 19), "J813715", "T1019", "2.32", "2.30")]
+    out = reconcile_visits_to_835(visits, svcs)
+    assert out[0].category == OutcomeCategory.NOT_IN_835
+    assert out[0].unpaid == Decimal("2.32")
+
+
+def test_partial_when_total_paid_below_95pct_of_billed():
+    visits = [_v(date(2026, 1, 18), "J813715", "T1019", "10.00")]
+    svcs = [_s(date(2026, 1, 18), "J813715", "T1019", "10.00", "4.00")]
+    out = reconcile_visits_to_835(visits, svcs)
+    assert out[0].category == OutcomeCategory.PARTIAL
+    assert out[0].unpaid == Decimal("6.00")
+
+
+def test_different_member_does_not_match():
+    visits = [_v(date(2026, 1, 18), "J813715", "T1019", "2.32")]
+    svcs = [_s(date(2026, 1, 18), "OTHERMEMBER", "T1019", "2.32", "2.30")]
+    out = reconcile_visits_to_835(visits, svcs)
+    assert out[0].category == OutcomeCategory.NOT_IN_835
diff --git a/backend/tests/test_rebill_summary.py b/backend/tests/test_rebill_summary.py
new file mode 100644
index 0000000..a78c1c8
--- /dev/null
+++ b/backend/tests/test_rebill_summary.py
@@ -0,0 +1,44 @@
+"""Summary CSV: one row per visit with its disposition."""
+from datetime import date
+from decimal import Decimal
+from pathlib import Path
+from cyclone.rebill.summary import (
+    SummaryRow,
+    write_summary_csv,
+    REBILLED_A, REBILLED_B, EXCLUDED_CARC, EXCLUDED_TIMELY_FILING,
+    EXCLUDED_PAYER, EXCLUDED_NO_EVV,
+)
+from cyclone.rebill.reconcile import VisitRow
+
+
+def test_summary_csv_emits_one_row_per_visit(tmp_path: Path):
+    out = tmp_path / "summary.csv"
+    rows = [
+        SummaryRow(
+            visit=VisitRow(date=date(2026, 6, 27), member_id="J813715",
+                           procedure="T1019", billed=Decimal("2.32")),
+            disposition=REBILLED_A, unpaid=Decimal("0.00"),
+            cas_reasons=("CO-45",), file_path="pipeline-a/x.837",
+        ),
+        SummaryRow(
+            visit=VisitRow(date=date(2026, 1, 1), member_id="OLD",
+                           procedure="T1019", billed=Decimal("2.32")),
+            disposition=EXCLUDED_TIMELY_FILING, unpaid=Decimal("2.32"),
+            cas_reasons=(), file_path="",
+        ),
+    ]
+    write_summary_csv(rows, out)
+    text = out.read_text()
+    assert "J813715" in text
+    assert "REBILLED_A" in text
+    assert "EXCLUDED_TIMELY_FILING" in text
+    assert "OLD" in text
+
+
+def test_disposition_constants():
+    assert REBILLED_A == "REBILLED_A"
+    assert REBILLED_B == "REBILLED_B"
+    assert EXCLUDED_CARC == "EXCLUDED_CARC"
+    assert EXCLUDED_PAYER == "EXCLUDED_PAYER"
+    assert EXCLUDED_NO_EVV == "EXCLUDED_NO_EVV"
+    assert EXCLUDED_TIMELY_FILING == "EXCLUDED_TIMELY_FILING"
diff --git a/backend/tests/test_rebill_timely_filing.py b/backend/tests/test_rebill_timely_filing.py
new file mode 100644
index 0000000..c8aeb87
--- /dev/null
+++ b/backend/tests/test_rebill_timely_filing.py
@@ -0,0 +1,39 @@
+"""120-day DOS-age gate. CO Medicaid's timely-filing window is 120 days from DOS."""
+from datetime import date
+from decimal import Decimal
+from cyclone.rebill.timely_filing import timely_filing_decision, FILING_WINDOW_DAYS
+
+
+def test_visit_within_window_is_rebillable():
+    decision = timely_filing_decision(
+        dos=date(2026, 6, 1), as_of=date(2026, 7, 7), override=False
+    )
+    assert decision.rebillable is True
+
+
+def test_visit_past_window_excluded_by_default():
+    decision = timely_filing_decision(
+        dos=date(2026, 1, 1), as_of=date(2026, 7, 7), override=False
+    )
+    assert decision.rebillable is False
+    assert decision.days_old == 187
+
+
+def test_visit_at_exact_120_days_inclusive():
+    decision = timely_filing_decision(
+        dos=date(2026, 3, 9), as_of=date(2026, 7, 7), override=False
+    )
+    # 2026-03-09 → 2026-07-07 = 120 days
+    assert decision.days_old == 120
+    assert decision.rebillable is True
+
+
+def test_override_flag_relaxes_the_gate():
+    decision = timely_filing_decision(
+        dos=date(2026, 1, 1), as_of=date(2026, 7, 7), override=True
+    )
+    assert decision.rebillable is True
+
+
+def test_window_constant_is_120():
+    assert FILING_WINDOW_DAYS == 120
diff --git a/backend/tests/test_submission_dedup.py b/backend/tests/test_submission_dedup.py
new file mode 100644
index 0000000..f8cf7f4
--- /dev/null
+++ b/backend/tests/test_submission_dedup.py
@@ -0,0 +1,38 @@
+"""Claim-id dedup at the SFTP pre-flight stage."""
+from datetime import datetime
+from cyclone.store.exceptions import DuplicateClaimError
+from cyclone.store.submission_dedup import (
+    record_submission,
+    check_duplicate,
+    DEFAULT_WINDOW_DAYS,
+)
+
+
+def test_check_duplicate_returns_none_for_first_submission(tmp_path):
+    db = tmp_path / "test.db"
+    assert check_duplicate("ORIG-001", db_url=f"sqlite:///{db}") is None
+
+
+def test_check_duplicate_raises_within_window(tmp_path):
+    db = tmp_path / "test.db"
+    record_submission("ORIG-002", submitted_at=datetime(2026, 6, 1), db_url=f"sqlite:///{db}")
+    try:
+        check_duplicate("ORIG-002", db_url=f"sqlite:///{db}",
+                       now=datetime(2026, 6, 15))
+    except DuplicateClaimError as e:
+        assert e.claim_id == "ORIG-002"
+        assert e.original_submission_at == datetime(2026, 6, 1)
+    else:
+        raise AssertionError("expected DuplicateClaimError")
+
+
+def test_check_duplicate_passes_after_window(tmp_path):
+    db = tmp_path / "test.db"
+    record_submission("ORIG-003", submitted_at=datetime(2026, 1, 1), db_url=f"sqlite:///{db}")
+    # 100 days later — past the 30-day default window
+    assert check_duplicate("ORIG-003", db_url=f"sqlite:///{db}",
+                          now=datetime(2026, 4, 11)) is None
+
+
+def test_default_window_is_30_days():
+    assert DEFAULT_WINDOW_DAYS == 30